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

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

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

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

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

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

始める前に

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

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

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

    API を有効にする

必要なロール

Dataproc でデータリネージを使用するために必要な権限を取得するには、管理者に Dataproc クラスタ VM サービス アカウントに対して次の 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.lineage.enabled クラスタ プロパティを true に設定して Dataproc クラスタを作成します。

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 はデータリネージ情報をキャプチャして 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 コンソールで表示できます。また、Data Lineage API から JSON データの形式で取得することもできます。

例:

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

リネージグラフの例

次のステップ