Managed Service for Apache Spark でデータ リネージを使用する

このドキュメントでは、Managed Service for Apache Spark バッチ ワークロードとインタラクティブ セッションで、プロジェクト、バッチ ワークロード、インタラクティブ セッションまたはランタイム プロファイル（セッション テンプレート）レベルでデータリネージを有効にする方法について説明します。

概要

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

Managed Service for Apache Spark のワークロードとセッションは、リネージ イベントをキャプチャして、Knowledge Catalog の Data Lineage API にパブリッシュします。Managed Service for Apache Spark は、OpenLineage Spark プラグインを使用して OpenLineage を介して Data Lineage API と統合されます。

Knowledge Catalog でリネージ情報にアクセスするには、リネージグラフと Data Lineage API を使用します。詳細については、Knowledge Catalog でリネージグラフを表示するをご覧ください。

対象

BigQuery と Cloud Storage のデータソースをサポートするデータリネージは、サポートされている Managed Service for Apache Spark ランタイム バージョンで実行されるワークロードとセッションで使用できます。ただし、次の例外と制限があります。

  • データリネージは、SparkR または Spark ストリーミングのワークロードまたはセッションでは使用できません。

始める前に

  1. Google Cloud コンソールのプロジェクト セレクタ ページで、Managed Service for Apache Spark のワークロードまたはセッションに使用するプロジェクトを選択します。

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

  2. データリネージ API を有効にします。

    API を有効にする

    Spark データリネージの今後の変更 Managed Service for Apache Spark のリリースノートで、Data Lineage API を有効にすると、プロジェクト、バッチ ワークロード、インタラクティブ セッションで Spark データリネージが自動的に利用可能になる変更のお知らせをご覧ください（サービスのデータリネージ取り込みを制御するをご覧ください）。追加のプロジェクト、バッチ ワークロード、インタラクティブ セッションの設定は必要ありません。

必要なロール

バッチ ワークロードでデフォルトの Managed Service for Apache Spark サービス アカウントを使用している場合、Dataproc Worker ロールが付与されています。このロールには datalineage.locations.processOpenLineageMessage 権限が含まれています。この権限は、OpenLineage Spark プラグインがリネージ イベントをキャプチャして Data Lineage API に公開するために使用します。追加のアクションは不要です。

バッチ ワークロードまたはインタラクティブ セッションでカスタム サービス アカウントを使用する場合は、サービス アカウントにリネージ イベントをパブリッシュする権限が必要です。最小権限の原則に従うには、ユースケースに必要な最小限のロールを付与します。

  • Dataproc ワーカー（roles/dataproc.worker）: Managed Service for Apache Spark ワークロードの実行に必要な標準ロール。このロールにはすでに datalineage.locations.processOpenLineageMessage 権限が含まれているため、カスタム サービス アカウントにこのロールがすでに付与されている場合は、追加のデータリネージ ロールを付与する必要はありません。
  • データリネージ イベント プロデューサー（roles/datalineage.producer）: カスタム サービス アカウントがリネージ イベントをパブリッシュするために必要な最小限のスタンドアロン データリネージ ロール。サービス アカウントに Dataproc ワーカーロールがない場合は、このロールを付与します。
  • データリネージ編集者（roles/datalineage.editor）: リネージ プロセスとイベントの公開に加えて、編集と削除を行う権限を付与します。このロールは、標準のパイプライン実行に必要な権限よりも多くの権限を付与します。このロールは、ワークロードで既存のリネージ メタデータを変更する必要がある場合にのみ必要です。
  • データリネージ管理者（roles/datalineage.admin）: データリネージ リソースに対する完全な管理アクセス権を付与します。このロールは、ワークロード サービス アカウントには必要ありません。

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

プロジェクト、バッチ ワークロード、インタラクティブ セッションで Spark データリネージを有効にできます。

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

プロジェクト レベルで Spark データリネージを有効にすると、バッチ ワークロードまたはインタラクティブ セッションで実行される後続の Spark ジョブで Spark データリネージが有効になります。

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

キー 値
DATAPROC_LINEAGE_ENABLED true

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

Spark バッチ ワークロードでデータリネージを有効にする

バッチ ワークロードでデータリネージを有効にするには、ワークロードを送信するときに spark.dataproc.lineage.enabled プロパティを true に設定します。この設定は、プロジェクト レベルの Spark データリネージ設定をオーバーライドします。プロジェクト レベルで Spark データリネージが無効になっているが、バッチ ワークロードで有効になっている場合、バッチ ワークロードの設定が優先されます。

Spark バッチ ワークロードで Spark データリネージを無効にするには、ワークロードを送信するときに spark.dataproc.lineage.enabled プロパティを false に設定します。

この例では、gcloud CLI を使用して、Spark リネージが有効になっているバッチ lineage-example.py ワークロードを送信します。

gcloud dataproc batches submit pyspark lineage-example.py \
    --region=REGION \
    --deps-bucket=gs://BUCKET \
    --properties=spark.dataproc.lineage.enabled=true

次の lineage-example.py コードは、一般公開の BigQuery テーブルからデータを読み取り、出力を既存の BigQuery データセットの新しいテーブルに書き込みます。一時ストレージには Cloud Storage バケットを使用します。

#!/usr/bin/env python

from pyspark.sql import SparkSession
import sys

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

source = 'bigquery-public-data:samples.shakespeare'
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_table = 'PROJECT_ID:DATASET.TABLE'
word_count.write.format('bigquery') \
  .option('table', destination_table) \
  .option('writeMethod', 'direct') \
  .save()

次のように置き換えます。

  • REGION: ワークロードを実行するリージョン
  • BUCKET: 依存関係を保存する既存の Cloud Storage バケットの名前
  • PROJECT_ID、DATASET、TABLE: プロジェクト ID、既存の BigQuery データセットの名前、データセットに作成する新しいテーブルの名前（テーブルが存在しない場合）

リネージグラフは、Knowledge Catalog UI で表示できます。

Spark リネージグラフ

Spark インタラクティブ セッションまたはランタイム プロファイル（セッション テンプレート）でデータリネージを有効にする

Spark インタラクティブ セッションまたはランタイム プロファイル（セッション テンプレート）でデータリネージを有効にするには、セッションまたはランタイム プロファイル（セッション テンプレート）を作成するときに、spark.dataproc.lineage.enabled プロパティを true に設定します。たとえば、 Google Cloud コンソールの [ランタイム プロファイルの作成] ページの [追加構成] で、[その他のカスタマイズ] を編集し、[Spark プロパティ] に値 true を持つ spark.dataproc.lineage.enabled プロパティを追加します。この設定は、プロジェクト レベルの Spark データリネージ設定をオーバーライドします。Spark データリネージがプロジェクト レベルで無効になっているが、インタラクティブ セッションで有効になっている場合、インタラクティブ セッションの設定が優先されます。

Spark インタラクティブ セッションまたはランタイム プロファイル（セッション テンプレート）で Spark データリネージを無効にするには、インタラクティブ セッションまたはランタイム プロファイル（セッション テンプレート）の作成時に spark.dataproc.lineage.enabled プロパティを false に設定します。

次の PySpark ノートブック コードは、Spark データリネージを有効にして Managed Service for Apache Spark インタラクティブ セッションを構成します。次に、Spark Connect セッションを作成します。このセッションでは、一般公開の BigQuery Shakespeare データセットに対してワードカウント クエリを実行し、出力を既存の BigQuery データセットの新しいテーブルに書き込みます（Spark Connect クライアントを使用するをご覧ください）。

# Configure the Dataproc Serverless interactive session
# to enable Spark data lineage.
from google.cloud.dataproc_v1 import Session

session = Session()
session.runtime_config.properties["spark.dataproc.lineage.enabled"] = "true"

# Create the Spark Connect session.
from google.cloud.managed_spark_connect import ManagedSparkSession

spark = (
    ManagedSparkSession.builder
    .sessionId("lineage-example")
    .sessionConfig(session)
    .getOrCreate()
)

# Run a wordcount query on the public BigQuery Shakespeare dataset.
source = "bigquery-public-data:samples.shakespeare"
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')

# Output the results to a BigQuery destination table.
destination_table = 'PROJECT_ID:DATASET.TABLE'
word_count.write.format('bigquery') \
  .option('table', destination_table) \
  .save()

次のように置き換えます。

  • PROJECT_ID、DATASET、TABLE: プロジェクト ID、既存の BigQuery データセットの名前、データセットに作成する新しいテーブルの名前（テーブルが存在しない場合）

データリネージ グラフを表示するには、BigQuery の [エクスプローラ] ページのナビゲーション パネルに表示されている宛先テーブル名をクリックし、テーブルの詳細ウィンドウで [リネージ] タブを選択します。

Spark リネージグラフ

Knowledge Catalog でリネージを表示する

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

次のステップ