Mengintegrasikan dengan OpenLineage

Dokumen ini menjelaskan cara mengintegrasikan OpenLineage dengan Knowledge Catalog (sebelumnya Dataplex Universal Catalog) untuk mengimpor dan memvisualisasikan silsilah data dari sistem eksternal. Dengan bertindak sebagai konsumen OpenLineage menggunakan REST API ProcessOpenLineageRunEvent, Knowledge Catalog memungkinkan Anda menyatukan silsilah pipeline kustom bersama silsilah bawaan dari layanan Google Cloud .

Ringkasan

OpenLineage adalah platform terbuka untuk mengumpulkan dan menganalisis informasi silsilah data. Dengan menggunakan standar terbuka untuk data silsilah, OpenLineage menangkap peristiwa silsilah dari komponen pipeline data yang menggunakan OpenLineage API untuk melaporkan operasi, tugas, dan set data.

Melalui Data Lineage API, Anda dapat mengimpor peristiwa OpenLineage untuk ditampilkan di antarmuka web Knowledge Catalog bersama informasi silsilah dari layananGoogle Cloud , seperti BigQuery, Managed Service untuk Apache Airflow, Cloud Data Fusion, dan Managed Service untuk Apache Spark.

Untuk mengimpor peristiwa OpenLineage yang menggunakan spesifikasi OpenLineage, gunakan metode REST API ProcessOpenLineageRunEvent, dan petakan faset OpenLineage ke atribut Data Lineage API.

Batasan integrasi OpenLineage

  • Versi yang didukung: Data Lineage API mendukung OpenLineage versi utama 1 dan 2.

  • Tindakan API: endpoint API Data Lineage ProcessOpenLineageRunEvent hanya bertindak sebagai konsumen pesan OpenLineage, bukan produsen. API ini memungkinkan Anda mengirim informasi silsilah yang dihasilkan oleh alat atau sistem yang mematuhi OpenLineage ke Knowledge Catalog. Beberapa layanan Google Cloud , seperti Managed Service untuk Apache Spark dan Managed Airflow, menyertakan produsen OpenLineage bawaan yang dapat mengirim peristiwa ke endpoint ini, sehingga mengotomatiskan pengambilan silsilah dari layanan tersebut.

  • Fitur yang tidak didukung: Data Lineage API tidak mendukung hal berikut:

    • Setiap rilis OpenLineage berikutnya dengan perubahan format pesan
    • DatasetEvent
    • JobEvent

  • Ukuran pesan: Ukuran maksimum satu pesan adalah 5 MB.

  • Panjang nama: Panjang setiap Nama yang Memenuhi Syarat dalam input dan output dibatasi hingga 4.000 karakter.

  • Silsilah tingkat kolom: Endpoint ProcessOpenLineageRunEvent membuat edge silsilah tingkat tabel hanya untuk peristiwa OpenLineage kustom. Grafik silsilah level kolom hanya dibuat untuk transformasi SQL BigQuery dan tugas Managed Service untuk Apache Spark. Facet tingkat kolom kustom tidak dirender di konsol.

  • Batas link: Link dikelompokkan menurut peristiwa, dengan maksimum 100 link per peristiwa. Jumlah agregat maksimum link tingkat tabel adalah 1.000. Untuk mesin otomatis yang didukung yang mengumpulkan silsilah tingkat kolom, jika pesan berisi lebih dari 1.500 link tingkat kolom, informasi tingkat kolom akan dilewati.

  • Cakupan grafik: Knowledge Catalog menampilkan grafik silsilah untuk setiap operasi tugas, yang menunjukkan input dan output peristiwa silsilah. Alat ini tidak mendukung proses tingkat rendah seperti tahap Spark.

Pemetaan atribut faset OpenLineage

Untuk mengetahui informasi tentang pemetaan OpenLineage, lihat Pemetaan OpenLineage.

Mengimpor peristiwa OpenLineage

Jika Anda belum menyiapkan OpenLineage, lihat Memulai.

Untuk mengimpor peristiwa OpenLineage ke Knowledge Catalog, panggil metode API ProcessOpenLineageRunEvent.

C#

C#

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan C# di Panduan memulai Data Lineage menggunakan library klien. Untuk mengetahui informasi selengkapnya, lihat dokumentasi referensi API C# Data Lineage.

Untuk melakukan autentikasi ke Data Lineage, siapkan Kredensial Default Aplikasi. Untuk mengetahui informasi selengkapnya, lihat Menyiapkan autentikasi untuk lingkungan pengembangan lokal. 

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

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan Go di Panduan memulai Data Lineage menggunakan library klien. Untuk mengetahui informasi selengkapnya, lihat dokumentasi referensi API Go Data Lineage.

Untuk melakukan autentikasi ke Data Lineage, siapkan Kredensial Default Aplikasi. Untuk mengetahui informasi selengkapnya, lihat Menyiapkan autentikasi untuk lingkungan pengembangan lokal. 


//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

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan Java di Panduan memulai Data Lineage menggunakan library klien. Untuk mengetahui informasi selengkapnya, lihat dokumentasi referensi API Java Data Lineage.

Untuk melakukan autentikasi ke Data Lineage, siapkan Kredensial Default Aplikasi. Untuk mengetahui informasi selengkapnya, lihat Menyiapkan autentikasi untuk lingkungan pengembangan lokal. 

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

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan Python di Panduan memulai Data Lineage menggunakan library klien. Untuk mengetahui informasi selengkapnya, lihat dokumentasi referensi API Python Data Lineage.

Untuk melakukan autentikasi ke Data Lineage, siapkan Kredensial Default Aplikasi. Untuk mengetahui informasi selengkapnya, lihat Menyiapkan autentikasi untuk lingkungan pengembangan lokal. 

# 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

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan Ruby di Panduan memulai Data Lineage menggunakan library klien. Untuk mengetahui informasi selengkapnya, lihat dokumentasi referensi API Ruby Data Lineage.

Untuk melakukan autentikasi ke Data Lineage, siapkan Kredensial Default Aplikasi. Untuk mengetahui informasi selengkapnya, lihat Menyiapkan autentikasi untuk lingkungan pengembangan lokal. 

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

Untuk mengimpor peristiwa OpenLineage, gunakan metode processOpenLineageRunEvent.

Sebelum menggunakan salah satu data permintaan, lakukan penggantian berikut:

  • PROJECT_ID: Project ID Google Cloud Anda.
  • LOCATION_ID: lokasi Google Cloud , seperti us-central1.

Metode HTTP dan URL: 

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

Meminta isi JSON: 

{
  "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"
}

Untuk mengirim permintaan Anda, perluas salah satu opsi berikut:

Anda akan melihat respons JSON seperti berikut:

{
  "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"
  ]
}

Alat untuk mengirim pesan OpenLineage

Untuk menyederhanakan pengiriman peristiwa ke Data Lineage API, Anda dapat menggunakan berbagai alat dan library:

  • Library klien Google Cloud untuk Data Lineage: Google menyediakan library klien untuk berinteraksi dengan Data Lineage API secara terprogram. Untuk petunjuk penginstalan, lihat Library klien.
  • Library Produsen Java Google Cloud: Google menyediakan library Java open source untuk membantu membuat dan mengirim peristiwa OpenLineage ke Data Lineage API. Untuk mengetahui informasi selengkapnya, lihat postingan blog Library java produsen untuk Silsilah Data kini bersifat open source. Library ini tersedia di GitHub dan Maven.
  • Transportasi GCP OpenLineage: Untuk produsen OpenLineage berbasis Java, tersedia GcpLineage Transport khusus. Library ini menyederhanakan integrasi dengan Data Lineage API, dengan meminimalkan kode yang diperlukan untuk mengirim peristiwa ke Data Lineage API. GcpLineageTransport dapat dikonfigurasi sebagai sink peristiwa untuk produsen OpenLineage yang ada, seperti Airflow, Spark, dan Flink. Untuk mengetahui informasi dan contoh selengkapnya, lihat GcpLineage.

Menganalisis informasi dari OpenLineage

Untuk menganalisis peristiwa OpenLineage yang diimpor, lihat Melihat grafik silsilah di UI Knowledge Catalog.

Data faset OpenLineage yang disimpan

Data Lineage API tidak menyimpan semua data faset dari pesan OpenLineage. Data Lineage API menyimpan kolom faset berikut:

  • spark_version
    • openlineage-spark-version
    • spark-version
  • semua spark.logicalPlan.*
  • environment-properties (facet silsilah Google Cloud kustom)
    • origin.sourcetype dan 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

Data Lineage API menyimpan informasi berikut:

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

Langkah berikutnya