Implementation:Datahub project Datahub OpenLineageToDataHub

Knowledge Sources	Datahub_project_Datahub
Domains	OpenLineage_Integration
Last Updated	2026-02-10 00:00 GMT

Overview

A comprehensive converter that transforms OpenLineage RunEvent objects into DataHub entities (DataFlow, DataJob, DataProcessInstance, Dataset) with support for dataset URN resolution, fine-grained column lineage, schema metadata, ownership, tags, and domains.

Description

OpenLineageToDataHub is a utility class (all static methods, private constructor) that serves as the primary bridge between the OpenLineage standard and DataHub's metadata model. At 1373 lines, it is the largest file in the openlineage-converter module and handles numerous conversion scenarios.

Dataset URN Resolution (convertOpenlineageDatasetToDatasetUrn()):

• Resolves symlinks (e.g., Glue catalog table references) to canonical dataset URNs
• Maps OpenLineage namespaces to DataHub platforms via the PLATFORM_MAP (e.g., awsathena to athena, sqlserver to mssql)
• Handles HDFS/S3 filesystem paths via HdfsPathDataset
• Supports platform instances, environment (FabricType), and URN aliasing
• Optionally lowercases URNs based on configuration

Run Event to Job Conversion (convertRunEventToJob()):

• Creates a DatahubJob with DataFlowInfo, DataJobInfo, DataProcessInstanceRunEvent, and DataProcessInstanceProperties
• Extracts custom properties from Spark, Airflow, and other processing engine facets
• Generates ownership from job facets
• Resolves parent job relationships

Fine-Grained Column Lineage (getFineGrainedLineage()):

• Extracts column-level lineage from ColumnLineageDatasetFacet
• Maps upstream/downstream schema fields to DataHub URNs
• Captures transformation operations and SQL queries from SQLJobFacet

MERGE INTO Handling (enhanced extraction):

• Detects MERGE INTO commands via job name patterns
• Extracts target table names from SQL facets, output dataset names, symlinks, or warehouse paths
• Appends table name to job URN for disambiguation

Orchestrator Detection (getOrchestrator()):

• Determines the orchestrator from processing engine, producer URL, or explicit configuration
• Recognizes Spark, Airflow, Trino, and OpenLineage producer patterns

Usage

Use this converter in any DataHub integration that receives OpenLineage events -- primarily the Spark lineage plugin and the Airflow OpenLineage integration. It is called by the event processing pipeline to transform raw OpenLineage events into DataHub MCPs for emission.

Code Reference

Source Location

• Repository: Datahub_project_Datahub
• File: metadata-integration/java/openlineage-converter/src/main/java/io/datahubproject/openlineage/converter/OpenLineageToDataHub.java

Signature

@Slf4j
public class OpenLineageToDataHub {

    // Dataset URN conversion
    public static Optional<DatasetUrn> convertOpenlineageDatasetToDatasetUrn(
        OpenLineage.Dataset dataset, DatahubOpenlineageConfig mappingConfig);

    // Run event conversion
    public static DatahubJob convertRunEventToJob(
        OpenLineage.RunEvent event, DatahubOpenlineageConfig datahubConf)
        throws IOException, URISyntaxException;

    // Flow utilities
    public static DataFlowUrn getFlowUrn(String namespace, String jobName,
        String processingEngine, URI producer, DatahubOpenlineageConfig config);
    public static DataFlowInfo convertRunEventToDataFlowInfo(
        OpenLineage.RunEvent event, String flowName);
    public static String getFlowName(String jobName, String flowName);

    // Metadata generation
    public static GlobalTags generateTags(List<String> tags);
    public static Domains generateDomains(List<String> domains);
    public static SchemaMetadata getSchemaMetadata(
        OpenLineage.Dataset dataset, DatahubOpenlineageConfig mappingConfig);
    public static SchemaFieldDataType.Type convertOlFieldTypeToDHFieldType(String openLineageFieldType);

    // Utility
    public static Edge createEdge(Urn urn, ZonedDateTime eventTime);
    public static AuditStamp createAuditStamp(ZonedDateTime eventTime);
}

Import

import io.datahubproject.openlineage.converter.OpenLineageToDataHub;

I/O Contract

Input	Type	Description
event	OpenLineage.RunEvent	OpenLineage run event with job, run, inputs, and outputs
datahubConf	DatahubOpenlineageConfig	Configuration for platform mapping, URN aliasing, lineage capture
dataset	OpenLineage.Dataset	Individual dataset with namespace and name

Output	Type	Description
DatahubJob	DatahubJob	Complete job model with flow, job, inputs, outputs, and process instance
DatasetUrn	Optional<DatasetUrn>	Resolved DataHub dataset URN
SchemaMetadata	SchemaMetadata	Schema fields from OpenLineage schema facet
GlobalTags	GlobalTags	Tag associations from Airflow DAG tags

Usage Examples

DatahubOpenlineageConfig config = DatahubOpenlineageConfig.builder()
    .fabricType(FabricType.PROD)
    .captureColumnLevelLineage(true)
    .materializeDataset(true)
    .build();

// Convert a full run event
DatahubJob job = OpenLineageToDataHub.convertRunEventToJob(runEvent, config);
List<MetadataChangeProposal> mcps = job.toMcps(config);

// Convert a single dataset URN
Optional<DatasetUrn> urn = OpenLineageToDataHub
    .convertOpenlineageDatasetToDatasetUrn(dataset, config);