自 2026 年 4 月 10 日起，Dataplex Universal Catalog 將更名為 Knowledge Catalog。API、用戶端程式庫、CLI 和 IAM 名稱維持不變。詳情請參閱「隆重推出 Google Cloud Knowledge Catalog」。

Google uses AI technology to translate content into your preferred language. AI translations can contain errors.

與 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。
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。
名稱長度：輸入和輸出內容中每個完整合格名稱的長度上限為 4000 個字元。
連結限制： 連結會依活動分組，每個活動最多可有 100 個連結。資料表層級連結的總數上限為 1000 個。如果郵件包含超過 1500 個資料欄層級的連結，系統會略過資料欄層級的資訊。
圖表範圍：Knowledge Catalog 會顯示每個工作執行的歷程圖表，並顯示歷程事件的輸入和輸出內容。不支援 Spark 階段等較低層級的程序。

OpenLineage Facet 屬性對應

如要瞭解 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

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 程式庫現已開放原始碼」網誌文章。這個程式庫可在 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 擷取及探索資料更新

除非另有註明，否則本頁面中的內容是採用創用 CC 姓名標示 4.0 授權，程式碼範例則為阿帕契 2.0 授權。詳情請參閱《Google Developers 網站政策》。Java 是 Oracle 和/或其關聯企業的註冊商標。

上次更新時間：2026-07-05 (世界標準時間)。