與 OpenLineage 整合

本文說明如何整合 OpenLineage 與 Knowledge Catalog (原稱 Dataplex Universal Catalog)，從外部系統匯入及顯示資料歷程。透過 ProcessOpenLineageRunEvent REST API 做為 OpenLineage 消費者，Knowledge Catalog 可讓您整合自訂管道歷程，以及 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。

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

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

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

• 郵件大小：單一郵件大小上限為 5 MB。

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

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

• 連結限制： 連結會依活動分組，每個活動最多可有 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

在試用這個範例之前，請先按照「使用用戶端程式庫的資料沿革快速入門導覽課程」中的 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 Library：Google 提供開放原始碼 Java 程式庫，協助建構 OpenLineage 事件並傳送至 Data Lineage API。詳情請參閱「資料沿襲的 Producer Java 程式庫現已開放原始碼」網誌文章。這個程式庫可在 GitHub 和 Maven 取得。
• OpenLineage GCP 傳輸：對於以 Java 為基礎的 OpenLineage 產生器，系統提供專用的 GcpLineage 傳輸。這項功能可減少將事件傳送至 Data Lineage API 時所需的程式碼，簡化與 Data Lineage API 的整合。GcpLineageTransport 可設為任何現有 OpenLineage 生產者 (例如 Airflow、Spark 和 Flink) 的事件接收器。如需更多資訊和範例，請參閱「GcpLineage」。

分析 OpenLineage 的資訊

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

儲存的 OpenLineage 層面資料

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

• 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 擷取及探索資料更新