Area: Product
Sub-Area: Metadata Management
Issue
When a structured property is deleted from DataHub but thousands of assets still have references to that property in their metadata, subsequent attempts to update structured properties on those assets will fail. This occurs because the aspect data contains an invalid structured property definition reference, creating corrupted metadata that prevents updates. This is often caused by a race condition during concurrent ingestion and deletion operations, where assets are being ingested with a structured property at the same time it's being deleted from the system.
Common Error Messages:
- "Failed to update structured property: Invalid aspect data"
- "Structured property definition not found"
- "Cannot set structured property on asset"
- Property update failures for specific entity types (e.g., Looker assets)
You Might Be Asking:
- Why can't I update structured properties on certain assets?
- How do I remove stale structured property references?
- What happens when I delete a structured property that's in use?
- Can I safely delete structured properties?
- Why are only some asset types affected (e.g., Looker but not others)?
Solution
Immediate Fix: Cleanup Script
Run a Python cleanup script to remove stale references from affected assets:
### Script to find a bad structured property and remove it
import os
import re
import json
from typing import Dict, List, Any
import logging
from dataclasses import dataclass
from avro.schema import RecordSchema
from importlib.metadata import version
from datahub.metadata.schema_classes import (
KEY_ASPECTS,
ASPECT_NAME_MAP,
StructuredPropertiesClass,
)
from datahub.ingestion.graph.client import DataHubGraph, get_default_graph
from datahub.emitter.mcp import MetadataChangeProposalWrapper
from datahub.emitter.serialization_helper import pre_json_transform
from datahub.configuration.common import OperationalError
from datahub.emitter.rest_emitter import EmitMode
from rich.progress import Progress
from rich.logging import RichHandler
logging.basicConfig(
level=logging.INFO, format="%(message)s", datefmt="[%X]", handlers=[RichHandler()]
)
logger = logging.getLogger("rich")
# Utility to get path of executed script regardless of where it is executed from
__location__ = os.path.realpath(os.path.join(os.getcwd(), os.path.dirname(__file__)))
@dataclass
class DataHubRegistryEntityEntry:
# Category of the DataHub Entity
category: str
# Name of the key aspect for the DataHub Entity
key: str
# List of aspects that make up the DataHub Entity
aspects: list[str]
@dataclass
class DataHubIndex:
# Index of entities that exist with their aspect names
registry: dict[str, DataHubRegistryEntityEntry]
# Schemas of aspects
schemas: dict[str, RecordSchema]
def load_entity_registry() - DataHubIndex:
registry: dict[str, DataHubRegistryEntityEntry] = {}
schemas: dict[str, RecordSchema] = {}
datahub_version = version("acryl-datahub")
logger.debug(f"Processing version {datahub_version}")
for key in KEY_ASPECTS:
if (
KEY_ASPECTS[key].ASPECT_INFO
and "keyForEntity" in KEY_ASPECTS[key].ASPECT_INFO
):
entity_name: str = KEY_ASPECTS[key].ASPECT_INFO["keyForEntity"]
key_aspect: str = KEY_ASPECTS[key].ASPECT_NAME
category: str = KEY_ASPECTS[key].ASPECT_INFO["entityCategory"]
aspects: list[str] = [key_aspect]
aspects.extend(KEY_ASPECTS[key].ASPECT_INFO["entityAspects"])
registry[entity_name] = DataHubRegistryEntityEntry(
category=category, key=key_aspect, aspects=aspects
)
# Load aspect schemas
for aspect_name in aspects:
if aspect_name not in schemas:
aspect = ASPECT_NAME_MAP.get(aspect_name)
if aspect:
schemas[aspect_name] = aspect.RECORD_SCHEMA
else:
logger.warning(
f"Aspect: {aspect_name} not found in ASPECT_NAME_MAP"
)
logger.info("Finished loading DataHub's Entity Registry")
return DataHubIndex(registry=registry, schemas=schemas)
def get_graphql_type_implementation_values(
client: DataHubGraph, type: str
) - list[str]:
"""Get all types that implement the Entity interface."""
query = f"""
{{
__type(name: "{type}") {{
name
enumValues {{
name
description
}}
}}
}}
"""
data = client.execute_graphql(query)
possible_values = []
for type_info in data["__type"]["enumValues"]:
possible_values.append(type_info["name"])
return sorted(possible_values)
def get_entities_with_aspect(index: DataHubIndex, aspect_name: str) - list[str]:
return [
key.lower()
for key in index.registry.keys()
if aspect_name in index.registry.get(key).aspects
]
def split_on_first_exception(text: str) - tuple[str, str]:
"""
Splits a string on the word boundary after the first occurrence of 'Exception'
within a larger word (like 'ValidationExceptionCollection').
Args:
text: The input string to split
Returns:
A tuple containing (part_before_and_including_exception, part_after_exception)
If 'Exception' is not found, returns (original_string, '')
"""
import re
# Find the first occurrence of 'Exception' as a substring (can be part of a larger word)
match = re.search(r"Exception", text)
if match:
# Get the end index of 'Exception'
exception_end_index = match.end()
# Find the end of the word containing 'Exception'
word_end_match = re.search(r"\W", text[exception_end_index:])
if word_end_match:
word_end_index = exception_end_index + word_end_match.start()
else:
word_end_index = len(text)
# Split the string at the end of the word
first_part = text[:word_end_index]
second_part = text[word_end_index:]
return first_part, second_part
# If 'Exception' is not found, return the original string and an empty string
return text, ""
def extract_urn(text: str) - str | None:
"""
Extracts a URN from a string.
Args:
text: The input string possibly containing a URN
Returns:
The extracted URN or None if no URN is found
Example:
extract_urn("Property: urn:li:structuredProperty:1a1383dd-4c03-4c8f-a917-5b982f0f9afc, value: {string=None} should be an urn)")
'urn:li:structuredProperty:1a1383dd-4c03-4c8f-a917-5b982f0f9afc'
"""
# Pattern to match URNs - looks for 'urn:' followed by alphanumeric characters,
# colons, hyphens and other allowed characters in URNs
urn_pattern = r"urn:[a-zA-Z0-9:_\-\.]+(?::[a-zA-Z0-9\-]+)+(?=[\s,\)\}]|$)"
match = re.search(urn_pattern, text)
if match:
return match.group(0)
return None
class DataHubExceptionParser:
"""Parser for DataHub exception strings to structured objects."""
def __init__(self):
self.patterns = {
"entity_aspect": r"EntityAspect:\(([^,]+),([^)]+)\)",
"exceptions": r"Exceptions:\s*\[(.*)\]",
"audit_stamp": r"auditStamp=\{([^}]*)\}",
"system_metadata": r"systemMetadata=\{([^}]*)\}",
"record_template": r"recordTemplate=\{([^}]*properties=\[[^\]]*\][^}]*)\}",
"properties": r"properties=\[([^\]]*)\]",
"property_block": r"\{([^}]*propertyUrn=[^}]*)\}",
"actor": r"actor=([^,\s}]+)",
"time": r"time=(\d+)",
"last_observed": r"lastObserved=(\d+)",
"aspect_modified": r"aspectModified=\{([^}]*)\}",
"change_type": r"changeType=([^,\s}]+)",
"entity_urn": r"entityUrn=([^,\s}]+)",
"aspect_name": r"aspectName=([^,\s}]+)",
"sub_type": r"subType=([^,\s}]+)",
"msg": r"msg=(.+?)(?=\)\]|$)",
"property_urn": r"propertyUrn=([^,\s}]+)",
"values": r"values=\[([^\]]*)\]",
"double_value": r"double=([0-9.]+)",
"string_value": r"string=([^,\s}]+)",
"last_modified": r"lastModified=\{([^}]*)\}",
"created": r"created=\{([^}]*)\}",
}
def parse(self, input_string: str) - Dict[str, Any]:
"""Parse DataHub exception string into structured object."""
result = {}
# Parse EntityAspect
entity_aspect = self._parse_entity_aspect(input_string)
if entity_aspect:
result["EntityAspect"] = entity_aspect
# Parse Exceptions
exceptions = self._parse_exceptions(input_string)
if exceptions:
result["Exceptions"] = exceptions
return result
def _parse_entity_aspect(self, text: str) - Dict[str, str]:
"""Parse EntityAspect section."""
match = re.search(self.patterns["entity_aspect"], text)
if match:
return {"urn": match.group(1).strip(), "aspect": match.group(2).strip()}
return {}
def _parse_exceptions(self, text: str) - List[Dict[str, Any]]:
"""Parse Exceptions section."""
match = re.search(self.patterns["exceptions"], text, re.DOTALL)
if not match:
return []
exception_content = match.group(1)
exception = {
"type": "AspectValidationException",
"item": self._parse_exception_item(exception_content),
"changeType": self._extract_field(
exception_content, "change_type", "UPSERT"
),
"entityUrn": self._extract_field(exception_content, "entity_urn", ""),
"aspectName": self._extract_field(exception_content, "aspect_name", ""),
"subType": self._extract_field(exception_content, "sub_type", ""),
"msg": self._extract_field(exception_content, "msg", "").strip(),
}
return [exception]
def _parse_exception_item(self, text: str) - Dict[str, Any]:
"""Parse exception item details."""
item = {
"changeType": self._extract_field(text, "change_type", "UPSERT"),
"auditStamp": self._parse_audit_stamp(text),
"systemMetadata": self._parse_system_metadata(text),
"recordTemplate": self._parse_record_template(text),
"aspectName": self._extract_field(text, "aspect_name", ""),
"urn": self._extract_field(text, "entity_urn", ""),
}
return item
def _parse_audit_stamp(self, text: str) - Dict[str, Any]:
"""Parse auditStamp section."""
match = re.search(self.patterns["audit_stamp"], text)
if match:
audit_content = match.group(1)
return {
"actor": self._extract_field(audit_content, "actor", ""),
"time": int(self._extract_field(audit_content, "time", "0")),
}
return {}
def _parse_system_metadata(self, text: str) - Dict[str, Any]:
"""Parse systemMetadata section."""
match = re.search(self.patterns["system_metadata"], text)
if not match:
return {}
sys_content = match.group(1)
metadata = {}
last_observed = self._extract_field(sys_content, "last_observed", "")
if last_observed:
metadata["lastObserved"] = int(last_observed)
aspect_modified_match = re.search(self.patterns["aspect_modified"], sys_content)
if aspect_modified_match:
aspect_content = aspect_modified_match.group(1)
metadata["aspectModified"] = {
"actor": self._extract_field(aspect_content, "actor", ""),
"time": int(self._extract_field(aspect_content, "time", "0")),
}
return metadata
def _parse_record_template(self, text: str) - Dict[str, Any]:
"""Parse recordTemplate section."""
match = re.search(self.patterns["record_template"], text, re.DOTALL)
if not match:
return {}
template_content = match.group(1)
properties = self._parse_properties(template_content)
return {"properties": properties}
def _parse_properties(self, text: str) - List[Dict[str, Any]]:
"""Parse properties array."""
properties_match = re.search(self.patterns["properties"], text, re.DOTALL)
if not properties_match:
return []
properties_content = properties_match.group(1)
property_blocks = re.findall(
self.patterns["property_block"], properties_content, re.DOTALL
)
properties = []
for block in property_blocks:
prop = self._parse_single_property(block)
if prop:
properties.append(prop)
return properties
def _parse_single_property(self, block: str) - Dict[str, Any] | None:
"""Parse a single property block."""
prop = {}
# Parse lastModified
last_modified_match = re.search(self.patterns["last_modified"], block)
if last_modified_match:
lm_content = last_modified_match.group(1)
prop["lastModified"] = {
"actor": self._extract_field(lm_content, "actor", ""),
"time": int(self._extract_field(lm_content, "time", "0")),
}
# Parse created
created_match = re.search(self.patterns["created"], block)
if created_match:
created_content = created_match.group(1)
prop["created"] = {
"actor": self._extract_field(created_content, "actor", ""),
"time": int(self._extract_field(created_content, "time", "0")),
}
# Parse values
values_match = re.search(self.patterns["values"], block)
if values_match:
values_content = values_match.group(1)
values = []
# Check for double value
double_match = re.search(self.patterns["double_value"], values_content)
if double_match:
values.append({"double": float(double_match.group(1))})
# Check for string value
string_match = re.search(self.patterns["string_value"], values_content)
if string_match:
values.append({"string": string_match.group(1)})
prop["values"] = values
# Parse propertyUrn
prop["propertyUrn"] = self._extract_field(block, "property_urn", "")
return prop if prop.get("propertyUrn") else None
def _extract_field(self, text: str, pattern_key: str, default: str = "") - str:
"""Extract field value using pattern."""
match = re.search(self.patterns[pattern_key], text)
return match.group(1).strip() if match else default
def to_json(self, input_string: str, indent: int = 2) - str:
"""Parse and return as pretty-printed JSON."""
parsed = self.parse(input_string)
return json.dumps(parsed, indent=indent)
## Load DataHub's Entity Registry
index: DataHubIndex = load_entity_registry()
# DataHub Exception Parser
parser = DataHubExceptionParser()
# Connect to the DataHub instance configured in your ~/.datahubenv file.
client: DataHubGraph = get_default_graph()
ingest_proposal_url = f"{client._gms_server}/aspects?action=ingestProposal"
entities_with_structured_properties = get_entities_with_aspect(
index, "structuredProperties"
)
graphql_entity_names = get_graphql_type_implementation_values(client, "EntityType")
logger.debug(f"GraphQL Entities: {graphql_entity_names}")
processed_entities: list[str] = [
name
for name in graphql_entity_names
if name.lower() in entities_with_structured_properties
]
logger.debug(f"Computed Entities: {processed_entities}")
dry_run: bool = False
urns = client.get_urns_by_filter(entity_types=processed_entities)
with Progress() as progress:
task = progress.add_task("Processing execution requests...")
for urn in urns:
progress.update(task,description=str(f"Processing: urn: {urn}"))
props: StructuredPropertiesClass | None = client.get_aspect(
urn, StructuredPropertiesClass
)
if not props:
logger.debug(f"{urn} has no structured properties, skipping")
continue
mcp = MetadataChangeProposalWrapper(
entityUrn=urn,
aspect=props,
)
mcp_obj = pre_json_transform(mcp.to_obj())
payload_dict = {"proposal": mcp_obj, "async": "false"}
payload = json.dumps(payload_dict)
urns_to_remove: list[str] = []
try:
response = client._emit_generic(ingest_proposal_url, payload)
logger.debug(f"No validation error found: {response}")
continue
except OperationalError as e:
reason, error = split_on_first_exception(e.message)
if "ValidationExceptionCollection" in reason:
output: dict[str, Any] = parser.parse(error)
if "Exceptions" in output.keys():
exceptions: list[dict] = output["Exceptions"]
for exception in exceptions:
if (
"type" in exception
and exception["type"] == "AspectValidationException"
):
if "aspectName" in exception and (
exception["aspectName"] == "structuredProperties"
or exception["aspectName"] == "'structuredProperties'"
):
bad_urn = extract_urn(exception["msg"])
if bad_urn:
urns_to_remove.append(bad_urn)
except Exception as e:
logger.error(f"Unknown error type: {type(e)}, {e}")
continue
# Continue to the next URN
# Remove properties with URNs in urns_to_remove
if urns_to_remove and hasattr(props, "properties"):
# Filter out properties where propertyUrn is in urns_to_remove
props.properties = [
prop
for prop in props.properties
if getattr(prop, "propertyUrn", None) not in urns_to_remove
]
if not dry_run:
client.emit(
MetadataChangeProposalWrapper(entityUrn=urn, aspect=props),
emit_mode=EmitMode.SYNC_WAIT,
)
logger.info(f"Fixed {urn} by removing {urns_to_remove}")
What Can and Cannot Be Changed
Once a structured property is created: - ✅ Can change: Display name, description, allowed values (if expanding list) - ❌ Cannot change: Data type, cardinality, applicable entity types - ⚠️ Risky: Deleting properties that are actively used
Additional Notes
- Always check for asset dependencies before deleting structured properties
- Race conditions are more likely in high-frequency ingestion environments
- Consider implementing a staging/approval workflow for structural changes
- Looker and other API-based sources may be more susceptible due to timing of API calls
- This issue affects thousands of assets simultaneously, requiring bulk remediation
- Regular audits can help identify orphaned references before they cause problems
- Future DataHub versions may include automatic cleanup for this scenario
Related Documentation
Related Tickets
5375, 5287
Tags:structured-properties,
metadata-corruption,
cleanup,
race-condition,
looker,
stale-references,
aspect-data,
bulk-update,
property-deletion,
data-quality