與 OpenLineage 整合

本文說明如何將 OpenLineage 與 Knowledge Catalog (原稱 Dataplex Universal Catalog) 整合，以便從外部系統匯入及視覺化呈現資料歷程。Knowledge Catalog 可透過 ProcessOpenLineageRunEvent REST API 充當 OpenLineage 消費者，讓您整合自訂管道歷程和 Google Cloud 服務的內建歷程。

總覽

OpenLineage 是開放式平台，用於收集及分析資料歷程資訊。OpenLineage 使用開放標準處理歷程資料，從資料管道元件擷取歷程事件，並使用 OpenLineage API 回報執行作業、工作和資料集。

您可以透過 Data Lineage API 匯入 OpenLineage 事件，並在 Knowledge Catalog 網頁介面中顯示，以及Google Cloud 服務 (例如 BigQuery、Managed Service for Apache Airflow、Cloud Data Fusion 和 Managed Service for Apache Spark) 的歷程資訊。

如要匯入使用 OpenLineage 規格的 OpenLineage 事件，請使用 ProcessOpenLineageRunEvent REST API 方法，並將 OpenLineage 屬性對應至 Data Lineage API 屬性。

OpenLineage 整合限制

• 支援的版本：Data Lineage API 支援 OpenLineage 主要版本 1 和 2。

• API 動作：Data Lineage API 端點 ProcessOpenLineageRunEvent 僅做為 OpenLineage 訊息的「消費者」，而非「生產者」。您可以透過 API，將任何符合 OpenLineage 規範的工具或系統產生的歷程資訊，傳送至 Knowledge Catalog。部分 Google Cloud 服務 (例如 Managed Service for Apache Spark 和 Managed Airflow) 內建 OpenLineage 產生器，可將事件傳送至這個端點，自動擷取這些服務的歷程。

• 不支援的功能：Data Lineage API 不支援下列功能：

  • 任何後續 OpenLineage 版本，且訊息格式有變更
  • DatasetEvent
  • JobEvent

• 訊息大小：單一訊息的大小上限為 5 MB。

• 名稱長度：輸入和輸出內容中，每個完整名稱的長度上限為 4,000 個字元。

• 資料欄層級歷程：ProcessOpenLineageRunEvent 端點只會為自訂 OpenLineage 事件具體化資料表層級歷程邊緣。系統只會為 BigQuery SQL 轉換和 Managed Service for Apache Spark 工作，產生資料欄層級歷程圖。控制台不會顯示自訂資料欄層級 facet。

• 連結限制： 連結會依事件分組，每個事件最多可有 100 個連結。資料表層級連結的匯總數量上限為 1,000 個。如果支援的自動化引擎會收集資料欄層級歷程，且訊息包含超過 1,500 個資料欄層級連結，系統會略過資料欄層級資訊。

• 圖表範圍：Knowledge Catalog 會顯示各項工作執行的歷程圖，呈現歷程事件的輸入和輸出內容。不支援 Spark 階段等較低層級的程序。

OpenLineage 屬性對應

如要瞭解 OpenLineage 對應，請參閱「OpenLineage 對應」。

匯入 OpenLineage 事件

如果尚未設定 OpenLineage，請參閱「入門指南」。

如要將 OpenLineage 事件匯入 Knowledge Catalog，請呼叫 API 方法 ProcessOpenLineageRunEvent。

C#

C#

在試用這個範例之前，請先按照「使用用戶端程式庫的 Data Lineage 快速入門導覽課程」中的 C# 設定操作說明進行操作。詳情請參閱 Data Lineage C# API 參考文件。

如要向 Data Lineage 進行驗證，請設定應用程式預設憑證。詳情請參閱「為本機開發環境設定驗證機制」。

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

在試用這個範例之前，請先按照「使用用戶端程式庫的 Data Lineage 快速入門導覽課程」中的 Go 設定操作說明進行操作。詳情請參閱 Data Lineage Go API 參考文件。

如要向 Data Lineage 進行驗證，請設定應用程式預設憑證。詳情請參閱「為本機開發環境設定驗證機制」。

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

在試用這個範例之前，請先按照「使用用戶端程式庫的 Data Lineage 快速入門導覽課程」中的 Java 設定操作說明進行操作。詳情請參閱 Data Lineage Java API 參考文件。

如要向 Data Lineage 進行驗證，請設定應用程式預設憑證。詳情請參閱「為本機開發環境設定驗證機制」。

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

在試用這個範例之前，請先按照「使用用戶端程式庫的 Data Lineage 快速入門導覽課程」中的 Python 設定操作說明進行操作。詳情請參閱 Data Lineage Python API 參考文件。

如要向 Data Lineage 進行驗證，請設定應用程式預設憑證。詳情請參閱「為本機開發環境設定驗證機制」。

# 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

在試用這個範例之前，請先按照「使用用戶端程式庫的 Data Lineage 快速入門導覽課程」中的 Ruby 設定操作說明進行操作。詳情請參閱 Data Lineage Ruby API 參考文件。

如要向 Data Lineage 進行驗證，請設定應用程式預設憑證。詳情請參閱「為本機開發環境設定驗證機制」。

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

如要匯入 OpenLineage 事件，請使用 processOpenLineageRunEvent 方法。

使用任何要求資料之前，請先修改下列項目的值：

• PROJECT_ID：您的 Google Cloud 專案 ID。
• LOCATION_ID： Google Cloud 位置，例如 us-central1。

HTTP 方法和網址：

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

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

請展開以下其中一個選項，以傳送要求：

curl (Linux、macOS 或 Cloud Shell)

注意：下列指令假設您已執行 gcloud init 或 gcloud auth login，透過使用者帳戶登入 gcloud CLI，或已使用 Cloud Shell 自動登入 gcloud CLI。您可以執行 gcloud auth list，查看目前使用的帳戶。

將要求主體儲存在名為 request.json 的檔案中，然後執行下列指令：

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)

注意：下列指令假設您已執行 gcloud init 或 gcloud auth login，透過使用者帳戶登入 gcloud CLI。您可以執行 gcloud auth list，查看目前使用的帳戶。

將要求主體儲存在名為 request.json 的檔案中，然後執行下列指令：

$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

您應該會收到如下的 JSON 回覆：

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

傳送 OpenLineage 訊息的工具

如要簡化將事件傳送至 Data Lineage API 的程序，可以使用各種工具和程式庫：

• Data Lineage 適用的 Google Cloud 用戶端程式庫：Google 提供用戶端程式庫，可透過程式輔助方式與 Data Lineage API 互動。如需安裝操作說明，請參閱「用戶端程式庫」。
• Google Cloud Java Producer 程式庫：Google 提供開放原始碼 Java 程式庫，協助建構 OpenLineage 事件並傳送至 Data Lineage API。詳情請參閱網誌文章「Producer java library for Data Lineage is now open source」(Data Lineage 的生產者 Java 程式庫現已開放原始碼)。這個程式庫可在 GitHub 和 Maven 取得。
• OpenLineage GCP Transport：如果是以 Java 為基礎的 OpenLineage 產生器，可以使用專屬的 GcpLineage Transport。這項功能可盡量減少將事件傳送至 Data Lineage API 所需的程式碼，簡化與 Data Lineage API 的整合。GcpLineageTransport 可設為任何現有 OpenLineage 產生器 (例如 Airflow、Spark 和 Flink) 的事件接收器。如需更多資訊和範例，請參閱 GcpLineage。

分析 OpenLineage 的資訊

如要分析匯入的 OpenLineage 事件，請參閱「在 Knowledge Catalog UI 中查看歷程圖」。

儲存的 OpenLineage 構面資料

Data Lineage API 不會儲存 OpenLineage 訊息中的所有面向資料。Data Lineage API 會儲存下列 facet 欄位：

• spark_version
  • openlineage-spark-version
  • spark-version
• 所有spark.logicalPlan.*
• environment-properties (自訂 Google Cloud 歷程面向)
  • origin.sourcetype 和 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 會儲存下列資訊：

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

後續步驟

• 進一步瞭解透過 Managed Service for Apache Spark 追蹤資料歷程，以及 Hive 資料歷程整合功能。
• 在互動式實驗室中試用： 使用資料歷程和 OpenLineage 擷取及探索資料更新