Time Travel and Historical Metadata Queries

Organizations need to query metadata as it existed at a specific point in time, track changes over time, or understand the state of data assets in the past. Understanding DataHub's temporal query capabilities is important.

You Might Be Asking:

• Can I see what a dataset looked like last month?
• How do I query historical metadata?
• Can I track metadata changes over time?

1. Query specific aspect version:

query getHistoricalMetadata {
  dataset(urn: "YOUR_URN") {
    # Get specific version of an aspect
    aspects(input: {
      aspectName: "datasetProperties"
      version: 5  # Specific version number
    }) {
      aspects {
        version
        created {
          time
          actor
        }
        aspect
      }
    }
  }
}

2. Query metadata at specific timestamp:

def get_metadata_at_timestamp(entity_urn, aspect_name, timestamp_ms):
    """
    Get metadata as it existed at a specific point in time
    """
    query = """
    query getAspectHistory($urn: String!, $aspectName: String!) {
      dataset(urn: $urn) {
        aspects(input: {
          aspectName: $aspectName
          maxCount: 100
        }) {
          aspects {
            version
            created {
              time
            }
            aspect
          }
        }
      }
    }
    """

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

    aspects = result['data']['dataset']['aspects']['aspects']

    # Find version active at timestamp
    relevant_version = None
    for aspect in sorted(aspects, key=lambda x: x['created']['time']):
        if aspect['created']['time'] <= timestamp_ms:
            relevant_version = aspect
        else:
            break

    return relevant_version

# Usage - get dataset properties as of January 1, 2025
timestamp = int(datetime(2025, 1, 1).timestamp() * 1000)
historical_props = get_metadata_at_timestamp(
    "urn:li:dataset:(urn:li:dataPlatform:snowflake,db.schema.table,PROD)",
    "datasetProperties",
    timestamp
)
print(f"Description on Jan 1: {historical_props['aspect']['description']}")

3. Track metadata changes over time:

def get_metadata_timeline(entity_urn, aspect_name, start_date, end_date):
    """
    Get all metadata changes in a date range
    """
    query = """
    query getAspectHistory($urn: String!, $aspectName: String!) {
      dataset(urn: $urn) {
        aspects(input: {aspectName: $aspectName, maxCount: 100}) {
          aspects {
            version
            created { time actor }
            aspect
          }
        }
      }
    }
    """

    result = graphql_query(query, {"urn": entity_urn, "aspectName": aspect_name})
    aspects = result['data']['dataset']['aspects']['aspects']

    start_ms = int(start_date.timestamp() * 1000)
    end_ms = int(end_date.timestamp() * 1000)

    timeline = []
    for aspect in aspects:
        timestamp = aspect['created']['time']
        if start_ms <= timestamp <= end_ms:
            timeline.append({
                'timestamp': datetime.fromtimestamp(timestamp / 1000),
                'actor': aspect['created']['actor'],
                'version': aspect['version'],
                'content': aspect['aspect']
            })

    return sorted(timeline, key=lambda x: x['timestamp'])

# Usage - get all description changes in Q1 2025
changes = get_metadata_timeline(
    entity_urn,
    "datasetProperties",
    datetime(2025, 1, 1),
    datetime(2025, 3, 31)
)

for change in changes:
    print(f"{change['timestamp']}: {change['actor']} changed description to: {change['content']['description']}")

4. Compare metadata between two points in time:

def compare_metadata(entity_urn, aspect_name, timestamp1, timestamp2):
    """
    Compare metadata between two timestamps
    """
    metadata1 = get_metadata_at_timestamp(entity_urn, aspect_name, timestamp1)
    metadata2 = get_metadata_at_timestamp(entity_urn, aspect_name, timestamp2)

    diff = {}

    # Compare description
    if metadata1['aspect']['description'] != metadata2['aspect']['description']:
        diff['description'] = {
            'old': metadata1['aspect']['description'],
            'new': metadata2['aspect']['description']
        }

    # Compare custom properties
    props1 = metadata1['aspect'].get('customProperties', {})
    props2 = metadata2['aspect'].get('customProperties', {})

    for key in set(list(props1.keys()) + list(props2.keys())):
        if props1.get(key) != props2.get(key):
            diff[f'customProperty.{key}'] = {
                'old': props1.get(key),
                'new': props2.get(key)
            }

    return diff

# Usage
diff = compare_metadata(
    entity_urn,
    "datasetProperties",
    int(datetime(2025, 1, 1).timestamp() * 1000),
    int(datetime(2025, 2, 1).timestamp() * 1000)
)
print("Changes detected:", json.dumps(diff, indent=2))

5. Visualize metadata evolution:

import matplotlib.pyplot as plt
from collections import Counter

def visualize_tag_evolution(entity_urn, start_date, end_date):
    """
    Visualize how tags changed over time
    """
    query = """
    query getTagHistory($urn: String!) {
      dataset(urn: $urn) {
        aspects(input: {aspectName: "globalTags", maxCount: 100}) {
          aspects {
            version
            created { time }
            aspect
          }
        }
      }
    }
    """

    result = graphql_query(query, {"urn": entity_urn})
    aspects = result['data']['dataset']['aspects']['aspects']

    # Track tag counts over time
    timeline = []
    for aspect in aspects:
        timestamp = datetime.fromtimestamp(aspect['created']['time'] / 1000)
        if start_date <= timestamp <= end_date:
            tags = [tag['tag'] for tag in aspect['aspect']['tags']]
            timeline.append((timestamp, len(tags)))

    # Plot
    dates, counts = zip(*sorted(timeline))
    plt.figure(figsize=(12, 6))
    plt.plot(dates, counts, marker='o')
    plt.title('Tag Count Evolution')
    plt.xlabel('Date')
    plt.ylabel('Number of Tags')
    plt.grid(True)
    plt.savefig('tag_evolution.png')

6. Restore previous metadata version:

def restore_metadata_version(entity_urn, aspect_name, version):
    """
    Restore a previous version of metadata
    """
    # Get the specific version
    historical_metadata = get_metadata_at_version(entity_urn, aspect_name, version)

    if not historical_metadata:
        print(f"Version {version} not found")
        return

    # Re-emit the historical version
    emitter = DatahubRestEmitter("http://localhost:8080")
    emitter.emit_mcp(
        entity_urn=entity_urn,
        aspect_name=aspect_name,
        aspect=historical_metadata['aspect']
    )

    print(f"Restored version {version} of {aspect_name}")

# Usage
restore_metadata_version(
    entity_urn,
    "datasetProperties",
    version=10  # Restore version 10
)

DataHub uses event sourcing - all versions are retained. Aspect versions increment monotonically. Time travel queries can be expensive for entities with many versions. Consider implementing caching for frequently accessed historical data. Retention policies may eventually prune old versions.

• Aspect Versioning
• Metadata Model