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 ProcessOpenLineageRunEvent REST API, Knowledge Catalog memungkinkan Anda menyatukan silsilah pipeline kustom bersama silsilah bawaan dari Google Cloud layanan.

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 proses, tugas, dan set data.

Melalui Data Lineage API, Anda dapat mengimpor peristiwa OpenLineage untuk ditampilkan di antarmuka web Knowledge Catalog bersama informasi silsilah dari Google Cloud layanan, 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 ProcessOpenLineageRunEvent REST API, dan petakan aspek OpenLineage ke atribut Data Lineage API.

Batasan integrasi OpenLineage

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

• 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 kompatibel dengan OpenLineage ke Knowledge Catalog. Beberapa Google Cloud layanan, seperti Managed Service untuk Apache Spark dan Managed Airflow, menyertakan OpenLineage produsen 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:

  • Rilis OpenLineage berikutnya dengan perubahan format pesan
  • DatasetEvent
  • JobEvent

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

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

• Batas link: Link dikelompokkan berdasarkan peristiwa, dengan maksimum 100 link per peristiwa. Jumlah agregat maksimum link tingkat tabel adalah 1.000. Jika pesan berisi lebih dari 1.500 link tingkat kolom, informasi tingkat kolom akan dilewati.

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

Pemetaan atribut aspek 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 Data Lineage C# API dokumentasi referensi.

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 Data Lineage Go API dokumentasi referensi.

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 Data Lineage Java API dokumentasi referensi.

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 Data Lineage Python API dokumentasi referensi.

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 Data Lineage Ruby API.

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 Anda Google Cloud .
• 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:

curl (Linux, macOS, atau Cloud Shell)

Catatan: Perintah berikut mengasumsikan bahwa Anda telah login ke gcloud CLI menggunakan akun pengguna Anda dengan menjalankan gcloud init atau gcloud auth login, atau dengan menggunakan Cloud Shell, yang secara otomatis membuat Anda login ke gcloud CLI. Anda dapat memeriksa akun yang saat ini aktif dengan menjalankan gcloud auth list.

Simpan isi permintaan dalam file bernama request.json, dan jalankan perintah berikut:

curl -X POST \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "Content-Type: application/json; charset=utf-8" \
     -d @request.json \
     "https://datalineage.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION_ID:processOpenLineageRunEvent"

PowerShell (Windows)

Catatan: Perintah berikut mengasumsikan bahwa Anda telah login ke gcloud CLI menggunakan akun pengguna Anda dengan menjalankan gcloud init atau gcloud auth login. Anda dapat memeriksa akun yang saat ini aktif dengan menjalankan gcloud auth list.

Simpan isi permintaan dalam file bernama request.json, dan jalankan perintah berikut:

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred" }

Invoke-WebRequest `
    -Method POST `
    -Headers $headers `
    -ContentType: "application/json; charset=utf-8" `
    -InFile request.json `
    -Uri "https://datalineage.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION_ID:processOpenLineageRunEvent" | Select-Object -Expand Content

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 mengetahui 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 Data Lineage kini open source. Library ini tersedia di GitHub dan Maven.
• Transportasi GCP OpenLineage: Untuk produsen OpenLineage berbasis Java, Transportasi GcpLineage khusus tersedia. Transportasi 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 aspek OpenLineage yang disimpan

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

• spark_version
  • openlineage-spark-version
  • spark-version
• semua spark.logicalPlan.*
• environment-properties (custom Google Cloud lineage aspek)
  • 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

• Pelajari lebih lanjut silsilah data dengan Managed Service untuk Apache Spark dan silsilah data Hive integrasi.
• Coba di lab interaktif: Mengambil dan Menjelajahi Pembaruan Data Dengan Silsilah Data dan OpenLineage