Integrate with OpenLineage

This document explains how to integrate OpenLineage with Knowledge Catalog (formerly Dataplex Universal Catalog) to visualize lineage data from various systems.

Overview

OpenLineage is an open platform for collecting and analyzing data lineage information. Using an open standard for lineage data, OpenLineage captures lineage events from data pipeline components which use an OpenLineage API to report on runs, jobs, and datasets.

Through the Data Lineage API, you can import OpenLineage events to display in the Knowledge Catalog web interface alongside lineage information from Google Cloud services, such as BigQuery, Managed Service for Apache Airflow, Cloud Data Fusion, and Managed Service for Apache Spark.

To import OpenLineage events that use the OpenLineage specification, use the ProcessOpenLineageRunEvent REST API method, and map OpenLineage facets to Data Lineage API attributes.

Limitations

  • The Data Lineage API supports OpenLineage major version 1.

  • The Data Lineage API endpoint ProcessOpenLineageRunEvent only acts as a consumer of OpenLineage messages, not a producer. The API lets you send lineage information generated by any OpenLineage-compliant tool or system into Knowledge Catalog. Some Google Cloud services, such as Managed Service for Apache Spark and Managed Airflow, include built-in OpenLineage producers that can send events to this endpoint, automating lineage capture from those services.

  • The Data Lineage API doesn't support the following:

    • Any subsequent OpenLineage release with message format changes
    • DatasetEvent
    • JobEvent
  • Maximum size of a single message is 5 MB.

  • Length of each Fully Qualified Name in inputs and outputs is limited to 4000 characters.

  • Links are grouped by events, with a maximum of 100 links per event. The maximum aggregate number of table-level links is 1000. If a message contains more than 1500 column-level links, the column-level information is skipped.

  • Knowledge Catalog displays a lineage graph for each job run, showing the inputs and outputs of lineage events. It doesn't support lower-level processes such as Spark stages.

OpenLineage mapping

For information about OpenLineage mapping, see OpenLineage mapping.

Import an OpenLineage event

If you haven't yet set up OpenLineage, see Getting started.

To import an OpenLineage event into Knowledge Catalog, call the API method ProcessOpenLineageRunEvent.

C#

C#

Before trying this sample, follow the C# setup instructions in the Knowledge Catalog quickstart using client libraries. For more information, see the Knowledge Catalog C# API reference documentation.

To authenticate to Knowledge Catalog, set up Application Default Credentials. For more information, see Set up authentication for a local development environment.

using Google.Cloud.DataCatalog.Lineage.V1;
using Google.Protobuf.WellKnownTypes;

public sealed partial class GeneratedLineageClientSnippets
{
    /// <summary>Snippet for ProcessOpenLineageRunEvent</summary>
    /// <remarks>
    /// This snippet has been automatically generated and should be regarded as a code template only.
    /// It will require modifications to work:
    /// - It may require correct/in-range values for request initialization.
    /// - It may require specifying regional endpoints when creating the service client as shown in
    ///   https://cloud.google.com/dotnet/docs/reference/help/client-configuration#endpoint.
    /// </remarks>
    public void ProcessOpenLineageRunEventRequestObject()
    {
        // Create client
        LineageClient lineageClient = LineageClient.Create();
        // Initialize request argument(s)
        ProcessOpenLineageRunEventRequest request = new ProcessOpenLineageRunEventRequest
        {
            Parent = "",
            OpenLineage = new Struct(),
        };
        // Make the request
        ProcessOpenLineageRunEventResponse response = lineageClient.ProcessOpenLineageRunEvent(request);
    }
}

Go

Go

Before trying this sample, follow the Go setup instructions in the Knowledge Catalog quickstart using client libraries. For more information, see the Knowledge Catalog Go API reference documentation.

To authenticate to Knowledge Catalog, set up Application Default Credentials. For more information, see Set up authentication for a local development environment.


//go:build examples

package main

import (
	"context"

	lineage "cloud.google.com/go/datacatalog/lineage/apiv1"
	lineagepb "cloud.google.com/go/datacatalog/lineage/apiv1/lineagepb"
)

func main() {
	ctx := context.Background()
	// This snippet has been automatically generated and should be regarded as a code template only.
	// It will require modifications to work:
	// - It may require correct/in-range values for request initialization.
	// - It may require specifying regional endpoints when creating the service client as shown in:
	//   https://pkg.go.dev/cloud.google.com/go#hdr-Client_Options
	c, err := lineage.NewClient(ctx)
	if err != nil {
		// TODO: Handle error.
	}
	defer c.Close()

	req := &lineagepb.ProcessOpenLineageRunEventRequest{
		// TODO: Fill request struct fields.
		// See https://pkg.go.dev/cloud.google.com/go/datacatalog/lineage/apiv1/lineagepb#ProcessOpenLineageRunEventRequest.
	}
	resp, err := c.ProcessOpenLineageRunEvent(ctx, req)
	if err != nil {
		// TODO: Handle error.
	}
	// TODO: Use resp.
	_ = resp
}

Java

Java

Before trying this sample, follow the Java setup instructions in the Knowledge Catalog quickstart using client libraries. For more information, see the Knowledge Catalog Java API reference documentation.

To authenticate to Knowledge Catalog, set up Application Default Credentials. For more information, see Set up authentication for a local development environment.

import com.google.cloud.datacatalog.lineage.v1.LineageClient;
import com.google.cloud.datacatalog.lineage.v1.ProcessOpenLineageRunEventRequest;
import com.google.cloud.datacatalog.lineage.v1.ProcessOpenLineageRunEventResponse;
import com.google.protobuf.Struct;

public class SyncProcessOpenLineageRunEvent {

  public static void main(String[] args) throws Exception {
    syncProcessOpenLineageRunEvent();
  }

  public static void syncProcessOpenLineageRunEvent() throws Exception {
    // This snippet has been automatically generated and should be regarded as a code template only.
    // It will require modifications to work:
    // - It may require correct/in-range values for request initialization.
    // - It may require specifying regional endpoints when creating the service client as shown in
    // https://cloud.google.com/java/docs/setup#configure_endpoints_for_the_client_library
    try (LineageClient lineageClient = LineageClient.create()) {
      ProcessOpenLineageRunEventRequest request =
          ProcessOpenLineageRunEventRequest.newBuilder()
              .setParent("parent-995424086")
              .setOpenLineage(Struct.newBuilder().build())
              .setRequestId("requestId693933066")
              .build();
      ProcessOpenLineageRunEventResponse response =
          lineageClient.processOpenLineageRunEvent(request);
    }
  }
}

Python

Python

Before trying this sample, follow the Python setup instructions in the Knowledge Catalog quickstart using client libraries. For more information, see the Knowledge Catalog Python API reference documentation.

To authenticate to Knowledge Catalog, set up Application Default Credentials. For more information, see Set up authentication for a local development environment.

# This snippet has been automatically generated and should be regarded as a
# code template only.
# It will require modifications to work:
# - It may require correct/in-range values for request initialization.
# - It may require specifying regional endpoints when creating the service
#   client as shown in:
#   https://googleapis.dev/python/google-api-core/latest/client_options.html
from google.cloud import datacatalog_lineage_v1


def sample_process_open_lineage_run_event():
    # Create a client
    client = datacatalog_lineage_v1.LineageClient()

    # Initialize request argument(s)
    request = datacatalog_lineage_v1.ProcessOpenLineageRunEventRequest(
        parent="parent_value",
    )

    # Make the request
    response = client.process_open_lineage_run_event(request=request)

    # Handle the response
    print(response)

Ruby

Ruby

Before trying this sample, follow the Ruby setup instructions in the Knowledge Catalog quickstart using client libraries. For more information, see the Knowledge Catalog Ruby API reference documentation.

To authenticate to Knowledge Catalog, set up Application Default Credentials. For more information, see Set up authentication for a local development environment.

require "google/cloud/data_catalog/lineage/v1"

##
# Snippet for the process_open_lineage_run_event call in the Lineage service
#
# This snippet has been automatically generated and should be regarded as a code
# template only. It will require modifications to work:
# - It may require correct/in-range values for request initialization.
# - It may require specifying regional endpoints when creating the service
# client as shown in https://cloud.google.com/ruby/docs/reference.
#
# This is an auto-generated example demonstrating basic usage of
# Google::Cloud::DataCatalog::Lineage::V1::Lineage::Client#process_open_lineage_run_event.
#
def process_open_lineage_run_event
  # Create a client object. The client can be reused for multiple calls.
  client = Google::Cloud::DataCatalog::Lineage::V1::Lineage::Client.new

  # Create a request. To set request fields, pass in keyword arguments.
  request = Google::Cloud::DataCatalog::Lineage::V1::ProcessOpenLineageRunEventRequest.new

  # Call the process_open_lineage_run_event method.
  result = client.process_open_lineage_run_event request

  # The returned object is of type Google::Cloud::DataCatalog::Lineage::V1::ProcessOpenLineageRunEventResponse.
  p result
end

REST

To import an OpenLineage event, use the processOpenLineageRunEvent method.

Before using any of the request data, make the following replacements:

  • PROJECT_ID: your Google Cloud project ID.
  • LOCATION_ID: the Google Cloud location, such as us-central1.

HTTP method and URL:

POST https://datalineage.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION_ID:processOpenLineageRunEvent

Request JSON body:

{
  "eventTime": "2023-04-04T13:21:16.098Z",
  "eventType": "COMPLETE",
  "inputs": [
    {
      "name": "somename",
      "namespace": "customnamespace"
    }
  ],
  "job": {
    "name": "somename",
    "namespace": "customnamespace"
  },
  "outputs": [
    {
      "name": "somename",
      "namespace": "customnamespace"
    }
  ],
  "producer": "someproducer",
  "run": {
    "runId": "somerunid"
  },
  "schemaURL": "https://openlineage.io/spec/1-0-5/OpenLineage.json#/$defs/RunEvent"
}

To send your request, expand one of these options:

You should receive a JSON response similar to the following:

{
  "process": "projects/my-project/locations/us-central1/processes/my-process",
  "run": "projects/my-project/locations/us-central1/processes/my-process/runs/my-run",
  "lineageEvents": [
    "projects/my-project/locations/us-central1/processes/my-process/runs/my-run/lineageEvents/my-lineage-event"
  ]
}

Tools for sending OpenLineage messages

To simplify sending events to the Data Lineage API, you can use various tools and libraries:

  • Google Cloud Java Producer Library: Google provides an open-source Java library to help construct and send OpenLineage events to the Data Lineage API. For more information, see the blog post Producer java library for Data Lineage is now open source. The library is available on GitHub and Maven.
  • OpenLineage GCP Transport: For Java-based OpenLineage producers, a dedicated GcpLineage Transport is available. It simplifies integration with Data Lineage API, by minimizing the code needed for sending events to Data Lineage API. The GcpLineageTransport can be configured as the event sink for any existing OpenLineage producer such as Airflow, Spark, and Flink. For more information and examples, see GcpLineage.

Analyze information from OpenLineage

To analyze the imported OpenLineage events, see View lineage graphs in Knowledge Catalog UI.

Stored data

The Data Lineage API doesn't store all facets data from the OpenLineage messages. The Data Lineage API stores the following facet fields:

  • spark_version
    • openlineage-spark-version
    • spark-version
  • all spark.logicalPlan.*
  • environment-properties (custom Google Cloud lineage facet)
    • origin.sourcetype and origin.name
    • spark.app.id
    • spark.app.name
    • spark.batch.id
    • spark.batch.uuid
    • spark.cluster.name
    • spark.cluster.region
    • spark.job.id
    • spark.job.uuid
    • spark.project.id
    • spark.query.node.name
    • spark.session.id
    • spark.session.uuid

The Data Lineage API stores the following information:

  • eventTime
  • run.runId
  • job.namespace
  • job.name

What's next