Dataproc でデータリネージを有効にする

このドキュメントでは、Dataproc Spark ジョブのデータリネージをプロジェクトまたはクラスタレベルで有効にする方法について説明します。

概要

データリネージは Dataplex の機能で、システム内でのデータの移動（データの送信元、データの通過先、データに適用される変換）を追跡できます。

データリネージは、Dataproc Compute Engine 2.0.74 以降と 2.1.22 以降のイメージで、SparkR を除くすべての Dataproc Spark ジョブで使用できます。また、BigQuery と Cloud Storage のデータソースをサポートしています。

Dataproc クラスタでこの機能を有効にすると、Dataproc Spark ジョブはデータリネージ イベントをキャプチャし、Dataplex Data Lineage API にパブリッシュします。Dataproc は、OpenLineage Spark プラグインを使用して OpenLineage を介して Data Lineage API と統合されます。

データリネージ情報には、以下を使用して Dataplex を介してアクセスできます。

制限事項

データリネージは、SparkR ジョブまたは Spark ストリーミング ジョブでは使用できません。

始める前に

  1. Google Cloud コンソールの [プロジェクト セレクタ] ページで、リネージを追跡する Dataproc クラスタを含むプロジェクトを選択します。

    プロジェクト セレクタに移動

  2. Data Lineage API と Data Catalog API を有効にします。

    API を有効にする

必要なロール

Dataproc でデータリネージを使用するために必要な権限を取得するには、管理者に Dataproc クラスタ VM サービス アカウントに対して次の IAM ロールを付与するよう依頼してください。

ロールの付与については、プロジェクト、フォルダ、組織へのアクセスを管理するをご覧ください。

必要な権限は、カスタムロールや他の事前定義ロールから取得することもできます。

プロジェクト レベルでデータリネージを有効にする

データリネージはプロジェクト レベルで有効にできます。プロジェクトでデータリネージが有効になった後に作成されたクラスタで実行されるサポートされている Spark ジョブでは、データリネージが有効になります。既存のクラスタ（プロジェクト レベルでデータリネージを有効にする前に作成されたクラスタ）で実行されるジョブでは、データリネージは有効になりません。

プロジェクト レベルでデータリネージを有効にする方法

プロジェクト レベルでデータリネージを有効にするには、次のカスタム プロジェクト メタデータを設定します。

キー 値
DATAPROC_LINEAGE_ENABLED true
DATAPROC_CLUSTER_SCOPES https://www.googleapis.com/auth/cloud-platform

DATAPROC_LINEAGE_ENABLED メタデータを false に設定すると、プロジェクト レベルでデータリネージを無効にできます。

クラスタレベルでデータリネージを有効にする

クラスタの作成時にデータリネージを有効にすると、クラスタに送信されるサポートされているすべての Spark ジョブでデータリネージが有効になります。

クラスタレベルでデータリネージを有効にする方法

クラスタでデータリネージを有効にするには、dataproc:dataproc.lineage.enabled クラスタ プロパティを true に設定して Dataproc クラスタを作成します。

2.0 イメージ バージョン クラスタ: データリネージには、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.extraListeners プロパティを渡すことで、特定のジョブでデータリネージを無効にできます。

有効にすると、クラスタでデータリネージを無効にすることはできません。すべてのクラスタジョブのデータリネージを排除するには、dataproc:dataproc.lineage.enabled プロパティなしでクラスタを再作成します。

Spark ジョブの送信

データリネージを有効にして作成された Dataproc クラスタで Spark ジョブを送信すると、Dataproc はデータリネージ情報をキャプチャして Data Lineage 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 でリネージグラフを表示する

リネージの可視化グラフには、プロジェクト リソースとそれらを作成したプロセスの関係が表示されます。データリネージ情報は、Google Cloud コンソールでグラフの可視化の形式で、または JSON データの形式で Data Lineage API から取得できます。

詳細については、Dataplex UI でリネージグラフを表示するをご覧ください。

例:

次の 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 UI に次のリネージグラフを作成します。

リネージグラフの例

次のステップ