"Managed Service for Apache Spark" is the new name for the product formerly known as "Dataproc on Compute Engine" (cluster deployment) and "Google Cloud Serverless for Apache Spark" (serverless deployment).

‫Google משתמשת בטכנולוגיית AI כדי לתרגם תוכן לשפה המועדפת עליך. בתרגומים כאלו עשויות להיות שגיאות.

שימוש בתיעוד מקורות נתונים עם Managed Service for Apache Spark

במאמר הזה מוסבר איך להפעיל את שושלת הנתונים בעומסי עבודה באצווה ובסשנים אינטראקטיביים ב-Managed Service for Apache Spark ברמת הפרויקט, עומס העבודה באצווה או הסשן האינטראקטיבי או פרופיל זמן הריצה (תבנית הסשן).

סקירה כללית

שושלת נתונים היא תכונה של Knowledge Catalog שמאפשרת לעקוב אחרי האופן שבו הנתונים עוברים במערכות: מאיפה הם מגיעים, לאן הם מועברים ואילו טרנספורמציות מוחלות עליהם.

עומסי עבודה וסשנים של Managed Service for Apache Spark מתעדים אירועי שושלת נתונים ומפרסמים אותם ב-Data Lineage API של Knowledge Catalog. ‫Managed Service for Apache Spark משתלב עם Data Lineage API באמצעות OpenLineage, בעזרת הפלאגין OpenLineage Spark.

אפשר לגשת למידע על שושלת הנתונים דרך Knowledge Catalog, באמצעות תרשימי שושלת ו-Data Lineage API. מידע נוסף זמין במאמר הצגת תרשימי שושלת ב-Knowledge Catalog.

זמינות

אפשר להשתמש במעקב אחר שושלת נתונים, שתומך במקורות נתונים של BigQuery ו-Cloud Storage, בעומסי עבודה ובסשנים שפועלים עם גרסאות נתמכות של סביבת זמן הריצה של Managed Service for Apache Spark, בכפוף לחריגות ולמגבלות הבאות:

שושלת הנתונים לא זמינה לסשנים או לעומסי עבודה של סטרימינג ב-SparkR או ב-Spark.

לפני שמתחילים

בדף לבחירת הפרויקט במסוף Google Cloud , בוחרים את הפרויקט שבו רוצים להשתמש בסשנים או בעומסי העבודה של Managed Service for Apache Spark.

כניסה לדף לבחירת הפרויקט
מפעילים את Data Lineage API.

הפעלת ממשקי ה-API

שינויים צפויים בשושלת הנתונים של Spark: בנתוני הגרסה של Managed Service for Apache Spark מופיעה הודעה על שינוי שיאפשר להשתמש באופן אוטומטי בשושלת הנתונים של Spark בפרויקטים, בעומסי עבודה באצווה ובסשנים אינטראקטיביים, כשמפעילים את Data Lineage API (ראו שליטה בהטמעת השושלת בשירות) בלי צורך בהגדרות נוספות של פרויקטים, עומסי עבודה באצווה או סשנים אינטראקטיביים.

התפקידים הנדרשים

אם עומס העבודה באצווה משתמש בחשבון השירות שמוגדר כברירת מחדל של Managed Service for Apache Spark, יש לו את התפקיד Managed Service for Apache Spark Worker, שמכיל את ההרשאות שנדרשות לשושלת נתונים.

עם זאת, אם עומס העבודה באצווה משתמש בחשבון שירות בהתאמה אישית כדי להפעיל את שושלת הנתונים, צריך להקצות לחשבון השירות בהתאמה אישית אחד מהתפקידים שמפורטים בפסקה הבאה, שמכילים את ההרשאות שנדרשות לשושלת הנתונים.

כדי לקבל את ההרשאות שדרושות לשימוש בשושלת נתונים עם Managed Service for Apache Spark, בקשו מהאדמין להקצות לכם את התפקידים הבאים ב-IAM בחשבון השירות בהתאמה אישית של עומס העבודה באצווה:

מקצים אחד מהתפקידים הבאים:
- ‫Managed Service for Apache Spark Worker (roles/dataproc.worker)
- עריכת שושלת נתונים (roles/datalineage.editor)
- יצירת שושלת נתונים (roles/datalineage.producer)
- אדמין של שושלת נתונים (roles/datalineage.admin)

להסבר על מתן תפקידים, ראו איך מנהלים את הגישה ברמת הפרויקט, התיקייה והארגון.

יכול להיות שאפשר לקבל את ההרשאות הנדרשות גם באמצעות תפקידים בהתאמה אישית או תפקידים מוגדרים מראש.

הפעלת שושלת נתונים ב-Spark

אתם יכולים להפעיל את Spark data lineage בפרויקט, בעומס העבודה באצווה או בסשן האינטראקטיבי.

הפעלת שושלת נתונים ברמת הפרויקט

אחרי שמפעילים את שושלת הנתונים של Spark ברמת הפרויקט, היא תופעל במשימות הבאות של Spark שפועלות בעומס עבודה באצווה או בסשן אינטראקטיבי.

כדי להפעיל את שושלת הנתונים של Spark בפרויקט, מגדירים את המטא-נתונים הבאים בהתאמה אישית בפרויקט:

מפתח	ערך
`DATAPROC_LINEAGE_ENABLED`	`true`

כדי להשבית את שושלת הנתונים של Spark בפרויקט, צריך להגדיר את המטא-נתונים של DATAPROC_LINEAGE_ENABLED ל-false.

הפעלת שושלת נתונים בעומס עבודה באצווה של Spark

כדי להפעיל את שושלת הנתונים בעומס עבודה באצווה, מגדירים את המאפיין spark.dataproc.lineage.enabled לערך true כששולחים את עומס העבודה. ההגדרה הזו מבטלת כל הגדרה של Spark data lineage ברמת הפרויקט: אם Spark data lineage מושבת ברמת הפרויקט אבל מופעל בעומס העבודה באצווה, ההגדרה של עומס העבודה באצווה מקבלת קדימות.

כדי להשבית את שושלת הנתונים של Spark בעומס עבודה באצווה של Spark, מגדירים את המאפיין spark.dataproc.lineage.enabled לערך false כששולחים את עומס העבודה.

בדוגמה הזו משתמשים ב-CLI של gcloud כדי לשלוח עומס עבודה של lineage-example.py באצווה, כשהשושלת של Spark מופעלת.

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: מזהה הפרויקט, השם של מערך נתונים קיים ב-BigQuery והשם של טבלה חדשה שרוצים ליצור במערך הנתונים (הטבלה לא יכולה להיות קיימת).

אפשר לראות את תרשים השושלת בממשק המשתמש של Knowledge Catalog.

תרשים שושלת היוחסין של Spark

הפעלת שושלת נתונים בסשן אינטראקטיבי של Spark או בפרופיל סביבת זמן ריצה (תבנית סשן)

כדי להפעיל את שושלת הנתונים בסשן אינטראקטיבי של Spark או בפרופיל זמן ריצה (תבנית סשן), מגדירים את המאפיין spark.dataproc.lineage.enabled לערך true כשיוצרים את הסשן או את פרופיל זמן הריצה (תבנית סשן). לדוגמה, בדף Create a runtime profile במסוף Google Cloud , בקטע Additional configuration, עורכים את Other customizations ומוסיפים את המאפיין spark.dataproc.lineage.enabled עם הערך true בקטע Spark properties. ההגדרה הזו מבטלת כל הגדרה של שושלת נתונים של Spark ברמת הפרויקט: אם שושלת הנתונים של Spark מושבתת ברמת הפרויקט אבל מופעלת בסשן האינטראקטיבי, ההגדרה של הסשן האינטראקטיבי מקבלת עדיפות.

כדי להשבית את שושלת הנתונים של Spark בסשן אינטראקטיבי או בפרופיל זמן ריצה (תבנית סשן) של Spark, מגדירים את המאפיין spark.dataproc.lineage.enabled לערך false כשיוצרים את הסשן האינטראקטיבי או את פרופיל זמן הריצה (תבנית הסשן).

קוד ה-notebook הבא ב-PySpark מגדיר סשן אינטראקטיבי של Managed Service for Apache Spark עם הפעלה של שושלת נתונים ב-Spark. לאחר מכן הוא יוצר סשן של Spark Connect שמריץ שאילתה של ספירת מילים במערך נתונים ציבורי של שייקספיר ב-BigQuery, ואז כותב את הפלט בטבלה חדשה במערך נתונים קיים ב-BigQuery (ראו יצירת סשן של Spark ב-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()