Metadata Versioning and Audit History

Organizations need to track who changed what metadata and when, view historical versions of metadata, and audit metadata changes for compliance. Understanding DataHub's versioning system is key for governance.

You Might Be Asking:

• How do I see metadata change history?
• Who changed the dataset description?
• Can I restore previous metadata versions?

1. Query metadata change history:

query getMetadataHistory {
  dataset(urn: "YOUR_URN") {
    # Get aspect history
    aspects(input: {
      aspectName: "datasetProperties"
      maxCount: 10
    }) {
      aspects {
        version
        created {
          time
          actor
        }
        aspect
      }
    }
  }
}

2. View change timeline:

query getChangeLog {
  # View all changes to an entity
  dataset(urn: "YOUR_URN") {
    # Description changes
    aspects(input: {aspectName: "datasetProperties"}) {
      aspects {
        version
        created {
          time
          actor
        }
      }
    }

    # Tag changes
    aspects(input: {aspectName: "globalTags"}) {
      aspects {
        version
        created {
          time
          actor
        }
      }
    }

    # Owner changes
    aspects(input: {aspectName: "ownership"}) {
      aspects {
        version
        created {
          time
          actor
        }
      }
    }
  }
}

3. Generate audit report:

# Create metadata change audit report
def generate_audit_report(dataset_urn, start_date, end_date):
    """
    Generate report of all metadata changes in date range
    """
    query = """
    query getAuditLog($urn: String!, $aspectName: String!) {
      dataset(urn: $urn) {
        aspects(input: {aspectName: $aspectName, maxCount: 100}) {
          aspects {
            version
            created {
              time
              actor
            }
            aspect
          }
        }
      }
    }
    """

    aspect_names = [
        "datasetProperties",
        "globalTags",
        "ownership",
        "glossaryTerms",
        "domains"
    ]

    changes = []

    for aspect_name in aspect_names:
        result = graphql_query(query, {
            "urn": dataset_urn,
            "aspectName": aspect_name
        })

        for aspect in result['data']['dataset']['aspects']['aspects']:
            timestamp = aspect['created']['time']

            if start_date <= timestamp <= end_date:
                changes.append({
                    'timestamp': timestamp,
                    'actor': aspect['created']['actor'],
                    'aspect': aspect_name,
                    'version': aspect['version'],
                    'content': aspect['aspect']
                })

    # Sort by timestamp
    changes.sort(key=lambda x: x['timestamp'])

    return changes

# Generate report
changes = generate_audit_report(
    dataset_urn,
    start_date="2025-01-01",
    end_date="2025-01-31"
)

for change in changes:
    print(f"{change['timestamp']}: {change['actor']} modified {change['aspect']}")

4. Track metadata modifications:

# Subscribe to metadata change log for real-time auditing
from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    'MetadataChangeLog_v1',
    bootstrap_servers=['kafka:9092'],
    value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)

for message in consumer:
    mcl = message.value

    print(f"""
    Change detected:
    Entity: {mcl['entityUrn']}
    Aspect: {mcl['aspectName']}
    Change Type: {mcl['changeType']}
    Actor: {mcl['systemMetadata']['lastObserved']}
    """)

    # Store in audit log
    store_audit_record(mcl)

5. Restore previous metadata version:

# Get previous version and re-emit
def restore_previous_version(entity_urn, aspect_name, version):
    """
    Restore a previous version of metadata
    """
    query = """
    query getSpecificVersion($urn: String!, $aspectName: String!) {
      dataset(urn: $urn) {
        aspects(input: {aspectName: $aspectName, maxCount: 100}) {
          aspects {
            version
            aspect
          }
        }
      }
    }
    """

    result = graphql_query(query, {
        "urn": entity_urn,
        "aspectName": aspect_name
    })

    # Find the specific version
    for aspect in result['data']['dataset']['aspects']['aspects']:
        if aspect['version'] == version:
            # Re-emit this version
            emitter.emit_mcp(
                entity_urn=entity_urn,
                aspect_name=aspect_name,
                aspect=aspect['aspect']
            )
            print(f"Restored version {version}")
            return

    print(f"Version {version} not found")

6. Enable comprehensive audit logging:

# In DataHub configuration
datahub-gms:
  env:
    # Enable audit logging
    - name: ENABLE_METADATA_AUDIT
      value: "true"

    # Retain audit history
    - name: METADATA_AUDIT_RETENTION_DAYS
      value: "365"

    # Detailed logging
    - name: METADATA_AUDIT_LEVEL
      value: "DETAILED"  # or "SUMMARY"

DataHub uses event sourcing - all metadata changes are stored. Aspect versions increment with each change. Audit logs can grow large - implement retention policies. Consider exporting audit data to external systems for long-term retention.

• Metadata Versioning
• Change Log