File-Based Lineage Ingestion Format Issues

File-based lineage ingestion is failing due to incorrect JSON/YAML format, invalid URNs, or missing required fields. This approach is commonly used for custom ETL tools or when programmatic emission is not feasible.

Common Error Messages:

• Invalid JSON format in lineage file
• Missing required field: downstream
• Invalid URN format
• Failed to parse lineage file

You Might Be Asking:

• What's the correct format for file-based lineage?
• How do I create URNs for file-based lineage?
• Can I include column-level lineage in files?

1. Basic table-level lineage format:

Example for Snowflake. This can be adapted for other data sources/connectors.

[
  {
    "entityType": "dataset",
    "entityUrn": "urn:li:dataset:(urn:li:dataPlatform:snowflake,db.schema.target_table,PROD)",
    "changeType": "UPSERT",
    "aspectName": "upstreamLineage",
    "aspect": {
      "json": {
        "upstreams": [
          {
            "auditStamp": {
              "time": 1642425600000,
              "actor": "urn:li:corpuser:etl_service"
            },
            "dataset": "urn:li:dataset:(urn:li:dataPlatform:snowflake,db.schema.source_table_1,PROD)",
            "type": "TRANSFORMED"
          },
          {
            "auditStamp": {
              "time": 1642425600000,
              "actor": "urn:li:corpuser:etl_service"
            },
            "dataset": "urn:li:dataset:(urn:li:dataPlatform:snowflake,db.schema.source_table_2,PROD)",
            "type": "TRANSFORMED"
          }
        ]
      }
    }
  }
]

2. Column-level (fine-grained) lineage format:

Example for Snowflake. This can be adapted for other data sources/connectors.

[
  {
    "entityType": "dataset",
    "entityUrn": "urn:li:dataset:(urn:li:dataPlatform:snowflake,db.schema.target_table,PROD)",
    "changeType": "UPSERT",
    "aspectName": "upstreamLineage",
    "aspect": {
      "json": {
        "upstreams": [...],
        "fineGrainedLineages": [
          {
            "upstreamType": "FIELD_SET",
            "upstreams": [
              "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snowflake,db.schema.source,PROD),source_col)"
            ],
            "downstreamType": "FIELD",
            "downstreams": [
              "urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:snowflake,db.schema.target,PROD),target_col)"
            ],
            "transformOperation": "TRANSFORM",
            "confidenceScore": 1.0
          }
        ]
      }
    }
  }
]

3. Ingest lineage from file:

# Using DataHub CLI
datahub ingest -c lineage-recipe.yml

# Recipe configuration
# lineage-recipe.yml
source:
  type: "file"
  config:
    filename: "lineage.json"

sink:
  type: "datahub-rest"
  config:
    server: "http://localhost:8080"

4. Generate lineage file programmatically:

Example for Snowflake. This can be adapted for other data sources/connectors.

import json
from datahub.emitter.mce_builder import make_dataset_urn, make_schema_field_urn

def create_lineage_file(source_tables, target_table, output_file):
    lineage = []

    for target in target_tables:
        lineage_entry = {
            "entityType": "dataset",
            "entityUrn": make_dataset_urn("snowflake", target),
            "changeType": "UPSERT",
            "aspectName": "upstreamLineage",
            "aspect": {
                "json": {
                    "upstreams": [
                        {
                            "dataset": make_dataset_urn("snowflake", src),
                            "type": "TRANSFORMED"
                        }
                        for src in source_tables
                    ]
                }
            }
        }
        lineage.append(lineage_entry)

    with open(output_file, 'w') as f:
        json.dump(lineage, f, indent=2)

# Usage
create_lineage_file(
    source_tables=["db.schema.source1", "db.schema.source2"],
    target_table="db.schema.target",
    output_file="lineage.json"
)

5. Validate lineage file before ingestion:

# Dry run to validate format
datahub ingest -c lineage-recipe.yml --dry-run

# Check for URN format
# URNs must follow pattern:
# urn:li:dataset:(urn:li:dataPlatform:PLATFORM,DATASET_NAME,ENV)

File-based lineage is useful for batch processing or integrating custom ETL tools. Ensure all referenced datasets exist in DataHub before emitting lineage. Use consistent URN formats across all lineage emissions.

• Lineage Ingestion
• File Source