在 Dataproc 中啟用 Spark 資料系譜

本文件說明如何在專案或叢集層級為 Dataproc Spark 工作啟用資料系譜。

資料歷程是 Dataplex Universal Catalog 的功能，可讓您追蹤資料在系統中的移動情形，包括來源、傳遞目的地和採用的轉換機制。

資料系譜可用於所有 Dataproc Spark 工作 (SparkR 和 Spark 串流工作除外)，並支援 BigQuery 和 Cloud Storage 資料來源。在 Compute Engine 2.0.74 以上版本、2.1.22 以上版本、2.2.50 以上版本的映像檔中，Dataproc 就包含這項功能。

在 Dataproc 叢集中啟用這項功能後，Dataproc Spark 工作會擷取資料歷程事件，並發布至 Dataplex Universal Catalog 的 Data Lineage API。Dataproc 會透過 OpenLineage 整合 Data Lineage API，並使用 OpenLineage Spark 外掛程式。

您可以透過 Dataplex Universal Catalog 存取資料沿革資訊，方法如下：

事前準備

  1. 在 Google Cloud 控制台的專案選擇器頁面中，選取包含您要追蹤系譜的 Dataproc 叢集的專案。

    前往專案選取器

  2. 啟用 Data Lineage API。

    啟用 API

必要的角色

如果您使用預設 VM 服務帳戶建立 Dataproc 叢集，該叢集就會具有 Dataproc Worker 角色，可啟用資料沿革。你不必採取其他行動。

不過，如果您建立的 Dataproc 叢集使用自訂服務帳戶，為了在叢集中啟用資料系譜，您必須將必要角色授予自訂服務帳戶，如後文所述。

如要取得使用 Dataproc 資料系譜所需的權限，請要求管理員授予您叢集的 自訂服務帳戶的下列 IAM 角色：

如要進一步瞭解如何授予角色，請參閱「管理專案、資料夾和機構的存取權」。

您或許還可透過自訂角色或其他預先定義的角色取得必要權限。

在專案層級啟用 Spark 資料歷程

您可以在專案層級啟用 Spark 資料系譜。在專案啟用資料系譜後，在叢集上執行的支援 Spark 工作會啟用資料系譜。請注意，在現有叢集上執行的工作 (在專案層級啟用資料系譜之前建立的叢集) 不會啟用資料系譜。

如何在專案層級啟用 Spark 資料系譜

如要在專案層級啟用 Spark 資料系譜，請設定下列自訂專案中繼資料：

鍵 值
DATAPROC_LINEAGE_ENABLED true
DATAPROC_CLUSTER_SCOPES https://www.googleapis.com/auth/cloud-platform

您可以將 DATAPROC_LINEAGE_ENABLED 中繼資料設為 false，在專案層級停用 Spark 資料系譜。

在叢集層級啟用 Spark 資料系譜

建立叢集時，您可以啟用 Spark 資料系譜，這樣提交至叢集的所有支援 Spark 工作都會啟用資料系譜。

如何在叢集層級啟用 Spark 資料系譜

如要在叢集中啟用 Spark 資料系譜，請建立 Dataproc 叢集，並將 dataproc:dataproc.lineage.enabled 叢集屬性設為 true。

2.0 映像檔版本叢集：Spark 資料系譜需要 Dataproc 叢集 VM 存取 cloud-platform 範圍。使用 2.1 以上版本的映像檔建立的 Dataproc 映像檔版本叢集已啟用 cloud-platform。如果您在建立叢集時指定 Dataproc 映像檔版本 2.0，請將範圍設為 cloud-platform。

gcloud CLI 範例：

gcloud dataproc clusters create CLUSTER_NAME \
    --project PROJECT_ID \
    --region REGION \
    --properties 'dataproc:dataproc.lineage.enabled=true'

在工作中停用 Spark 資料系譜

如果您在叢集層級啟用 Spark 資料系譜，則可在提交工作時傳遞空值 spark.extraListeners 屬性 (「"」)，藉此在特定工作中停用 Spark 資料系譜。

啟用後，您就無法在叢集中停用 Spark 資料系譜。如要消除所有叢集工作中的 Spark 資料系譜，您可以重新建立叢集，但不使用 dataproc:dataproc.lineage.enabled 屬性。

提交 Spark 工作

在啟用 Spark 資料系譜功能的 Dataproc 叢集中提交 Spark 工作時，Dataproc 會擷取資料系譜資訊並回報至資料系譜 API。

gcloud dataproc jobs submit spark \
    --cluster=CLUSTER_NAME \
    --project PROJECT_ID \
    --region REGION \
    --class CLASS \
    --jars=gs://APPLICATION_BUCKET/spark-application.jar \
    --properties=spark.openlineage.namespace=CUSTOM_NAMESPACE,spark.openlineage.appName=CUSTOM_APPNAME

注意：

  • 您可以選擇新增 spark.openlineage.namespace 和 spark.openlineage.appName 屬性，用於識別工作。如果您未加入這些屬性，Dataproc 會使用下列預設值：
    • spark.openlineage.namespace 的預設值：PROJECT_ID
    • spark.openlineage.appName 的預設值：spark.app.name

在 Dataplex Universal Catalog 中查看歷程

歷程圖會顯示專案資源與建立這些資源的程序之間的關係。您可以在 Google Cloud 控制台查看資料沿革資訊，或以 JSON 資料格式從 Data Lineage API 擷取資料沿革資訊。

範例：

下列 Spark 工作會從 BigQuery 資料表讀取資料，然後寫入另一個 BigQuery 資料表。

#!/usr/bin/env python

from pyspark.sql import SparkSession
import sys

spark = SparkSession \
  .builder \
  .appName('LINEAGE_BQ_TO_BQ') \
  .getOrCreate()

bucket = lineage-ol-test
spark.conf.set('temporaryGcsBucket', bucket)

source = sample.source
words = spark.read.format('bigquery') \
  .option('table', source) \
  .load()
words.createOrReplaceTempView('words')

word_count = spark.sql('SELECT word, SUM(word_count) AS word_count FROM words GROUP BY word')

destination = sample.destination
word_count.write.format('bigquery') \
  .option('table', destination) \
  .save()

Spark 工作會在 Dataplex Universal Catalog UI 中建立下列歷程圖表：

歷程圖範例

後續步驟