Metadata Propagation Through Complex Lineage

Metadata (tags, owners, glossary terms) needs to propagate through multi-hop lineage relationships, but the propagation is not working as expected or is creating unintended associations.

You Might Be Asking:

• How far does metadata propagate through lineage?
• Can I control propagation distance?
• Why is metadata appearing on unrelated datasets?

1. Understand default propagation behavior:

By default:
- Tags: No automatic propagation (must be enabled per tag)
- Owners: No propagation
- Glossary Terms: No propagation
- Domains: No propagation

Propagation must be explicitly configured.

2. Enable tag propagation with hop limit:

mutation configureTagPropagation {
  updateTag(
    urn: "urn:li:tag:PII"
    input: {
      name: "PII"
      # Enable propagation
      propagate: true

      # Propagation settings
      propagateDownstream: true
      propagateUpstream: false

      # Limit propagation hops (custom implementation)
      # This requires custom policy or implementation
    }
  )
}

3. Implement custom propagation logic:

# Custom script to propagate metadata with hop control
from datahub.emitter.rest_emitter import DatahubRestEmitter
from datahub.metadata.schema_classes import TagAssociationClass, GlobalTagsClass

def propagate_tags_with_limit(
    emitter,
    source_urn,
    tag_urns,
    max_hops=2,
    direction="downstream"
):
    """
    Propagate tags through lineage with hop limit
    """
    visited = set()
    queue = [(source_urn, 0)]  # (urn, current_hop)

    while queue:
        current_urn, hop = queue.pop(0)

        if hop > max_hops or current_urn in visited:
            continue

        visited.add(current_urn)

        # Apply tags to current entity
        if hop > 0:  # Don't re-tag source
            apply_tags(emitter, current_urn, tag_urns)

        # Get lineage and continue propagation
        lineage = get_lineage(emitter, current_urn, direction)
        for related_urn in lineage:
            queue.append((related_urn, hop + 1))

def apply_tags(emitter, entity_urn, tag_urns):
    """Apply tags to entity"""
    tags = GlobalTagsClass(
        tags=[TagAssociationClass(tag=tag) for tag in tag_urns]
    )
    emitter.emit_mcp(
        entity_urn=entity_urn,
        aspect_name="globalTags",
        aspect=tags
    )

4. Query lineage depth:

query getMultiHopLineage {
  # Get 1-hop upstream
  dataset(urn: "YOUR_URN") {
    upstreamLineage {
      entities {
        urn
        name
        # Get 2-hop upstream
        upstreamLineage {
          entities {
            urn
            name
          }
        }
      }
    }
  }
}

5. Visualize propagation impact:

Before propagating:
1. Query the full lineage graph
2. Identify all assets that would be affected
3. Review if propagation makes sense for each asset
4. Consider excluding certain paths or asset types

Use case example:
- PII tag on source table → should propagate to derived tables
- But NOT to aggregated summary tables (PII no longer present)
- Implement conditional logic based on transformation type

6. Implement propagation policies:

# Policy-based propagation rules
propagation_rules:
  - tag: "urn:li:tag:PII"
    direction: "downstream"
    max_hops: 2
    exclude_patterns:
      - ".*_aggregated$"
      - ".*_summary$"

  - tag: "urn:li:tag:Financial"
    direction: "both"
    max_hops: 3
    require_transformation_type: ["COPY", "VIEW"]

Unbounded propagation can lead to metadata sprawl and incorrect classifications. Always test propagation rules on a subset before applying broadly. Consider transformation types when deciding propagation - aggregations may remove PII, while copies preserve it.

• Lineage Impact Analysis
• Tag Propagation