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 merekam peristiwa silsilah dari komponen pipeline data yang menggunakan OpenLineage API untuk melaporkan eksekusi, 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 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 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 produser 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:

  • 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 Sepenuhnya dalam input dan output dibatasi hingga 4.000 karakter.

• Batas link: Link dikelompokkan menurut acara, dengan maksimum 100 link per acara. Jumlah gabungan 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 eksekusi tugas, yang menunjukkan input dan output peristiwa silsilah. Tidak mendukung proses tingkat yang lebih 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 C# API 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 Go API 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 Java API 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 Python API 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 Ruby API 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: Google Cloud Project ID 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:

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 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 Producer java library for Data Lineage is now open source. Library ini tersedia di GitHub dan Maven.
• Transportasi OpenLineage GCP: Untuk produsen OpenLineage berbasis Java, tersedia GcpLineage Transport khusus. Hal 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 produser 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 (aspek 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

• Pelajari lebih lanjut asal-usul data dengan Managed Service untuk Apache Spark dan integrasi asal-usul data Hive.
• Coba di lab interaktif: Merekam dan Menjelajahi Pembaruan Data dengan Silsilah Data dan OpenLineage