Menggunakan silsilah data dengan Serverless untuk Apache Spark

Dokumen ini menjelaskan cara mengaktifkan asal data di Google Cloud Serverless untuk workload batch Apache Spark dan sesi interaktif di tingkat project, workload batch, atau sesi interaktif.

Ringkasan

Silsilah data adalah fitur Dataplex Universal Catalog yang memungkinkan Anda melacak bagaimana data berpindah melalui sistem: dari mana data berasal, ke mana data diteruskan, dan transformasi yang diterapkan pada data.

Google Cloud Workload dan sesi Serverless untuk Apache Spark merekam peristiwa silsilah dan memublikasikannya ke Data Lineage API Dataplex Universal Catalog. Serverless for Apache Spark terintegrasi dengan Data Lineage API melalui OpenLineage, menggunakan plugin OpenLineage Spark.

Anda dapat mengakses informasi silsilah melalui Dataplex Universal Catalog, menggunakan grafik silsilah dan Data Lineage API. Untuk mengetahui informasi selengkapnya, lihat Melihat grafik silsilah di Dataplex Universal Catalog.

Ketersediaan

Silsilah data, yang mendukung sumber data BigQuery dan Cloud Storage, tersedia untuk beban kerja dan sesi yang berjalan dengan versi runtime Serverless untuk Apache Spark yang didukung dengan pengecualian dan batasan berikut:

Silsilah data tidak tersedia untuk beban kerja atau sesi SparkR atau Spark streaming.

Sebelum memulai

Di halaman pemilih project di konsol Google Cloud , pilih project yang akan digunakan untuk beban kerja atau sesi Serverless untuk Apache Spark Anda.

Buka pemilih project
Aktifkan Data Lineage API.

Aktifkan API

Perubahan silsilah data Spark mendatang Lihat catatan rilis Serverless for Apache Spark untuk mengetahui pengumuman perubahan yang akan secara otomatis membuat silsilah data Spark tersedia untuk project, workload batch, dan sesi interaktif Anda saat Anda mengaktifkan Data Lineage API (lihat Mengontrol penyerapan silsilah untuk layanan) tanpa memerlukan setelan project, workload batch, atau sesi interaktif tambahan.

Peran yang diperlukan

Jika workload batch Anda menggunakan akun layanan Serverless for Apache Spark default, workload tersebut memiliki peran Dataproc Worker, yang berisi izin yang diperlukan oleh silsilah data.

Namun, jika workload batch Anda menggunakan akun layanan kustom untuk mengaktifkan asal data, Anda harus memberikan salah satu peran yang tercantum dalam paragraf berikut, yang berisi izin yang diperlukan oleh asal data, ke akun layanan kustom.

Untuk mendapatkan izin yang diperlukan untuk menggunakan silsilah data dengan Dataproc, minta administrator untuk memberi Anda peran IAM berikut di akun layanan kustom beban kerja batch Anda:

Beri salah satu peran berikut:
- Worker Dataproc (roles/dataproc.worker)
- Editor urutan data (roles/datalineage.editor)
- Produsen silsilah data (roles/datalineage.producer)
- Administrator urutan data (roles/datalineage.admin)

Untuk mengetahui informasi selengkapnya tentang pemberian peran, lihat Mengelola akses ke project, folder, dan organisasi.

Anda mungkin juga bisa mendapatkan izin yang diperlukan melalui peran khusus atau peran bawaan lainnya.

Mengaktifkan silsilah data Spark

Anda dapat mengaktifkan asal-usul data Spark untuk project, workload batch, atau sesi interaktif.

Mengaktifkan silsilah data di tingkat project

Setelah Anda mengaktifkan silsilah data Spark di tingkat project, tugas Spark berikutnya yang berjalan dalam workload batch atau sesi interaktif akan mengaktifkan silsilah data Spark.

Untuk mengaktifkan silsilah data Spark di project Anda, tetapkan metadata project kustom berikut:

Kunci	Nilai
`DATAPROC_LINEAGE_ENABLED`	`true`

Anda dapat menonaktifkan silsilah data Spark untuk project dengan menyetel metadata DATAPROC_LINEAGE_ENABLED ke false.

Mengaktifkan silsilah data untuk workload batch Spark

Untuk mengaktifkan silsilah data pada beban kerja batch, tetapkan properti spark.dataproc.lineage.enabled ke true saat Anda mengirimkan beban kerja. Setelan ini menggantikan setelan silsilah data Spark di tingkat project: jika silsilah data Spark dinonaktifkan di tingkat project, tetapi diaktifkan untuk workload batch, setelan workload batch akan lebih diutamakan.

Contoh ini menggunakan gcloud CLI untuk mengirimkan workload batch lineage-example.py dengan lineage Spark diaktifkan.

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

Kode lineage-example.py berikut membaca data dari tabel BigQuery publik, lalu menulis output ke tabel baru dalam set data BigQuery yang ada. Bucket ini menggunakan bucket Cloud Storage untuk penyimpanan sementara.

#!/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()

Ganti kode berikut:

REGION: Region untuk menjalankan workload
BUCKET: Nama bucket Cloud Storage yang ada untuk menyimpan dependensi
PROJECT_ID, DATASET, dan TABLE: ID project, nama set data BigQuery yang ada, dan nama tabel baru yang akan dibuat di set data (tabel tidak boleh ada)

Anda dapat melihat grafik silsilah di UI Dataplex Universal Catalog.

Grafik silsilah Spark

Mengaktifkan silsilah data untuk sesi interaktif Spark

Untuk mengaktifkan asal data pada sesi interaktif Spark, tetapkan properti spark.dataproc.lineage.enabled ke true saat Anda membuat sesi atau template sesi. Setelan ini menggantikan setelan silsilah data Spark di tingkat project: jika silsilah data Spark dinonaktifkan di tingkat project, tetapi diaktifkan untuk sesi interaktif, setelan sesi interaktif akan lebih diutamakan.

Kode notebook PySpark berikut mengonfigurasi sesi interaktif Serverless for Apache Spark dengan pelacakan asal data Spark diaktifkan. Kemudian, kode ini membuat sesi Spark Connect yang menjalankan kueri jumlah kata pada set data Shakespeare BigQuery publik, lalu menulis output ke tabel baru dalam set data BigQuery yang ada (lihat Membuat sesi Spark di notebook BigQuery Studio) .

# 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.dataproc_spark_connect import DataprocSparkSession

spark = DataprocSparkSession.builder.dataprocSessionConfig(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()