"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, יש לו את התפקיד Dataproc Worker. התפקיד הזה כולל את ההרשאה datalineage.locations.processOpenLineageMessage, שבה הפלאגין OpenLineage Spark משתמש כדי לתעד ולפרסם אירועי שושלת נתונים ב-Data Lineage API. אין צורך לבצע פעולה נוספת.

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

‫Dataproc Worker (roles/dataproc.worker): התפקיד הרגיל שנדרש כדי להריץ עומסי עבודה ב-Managed Service for Apache Spark. התפקיד הזה כבר כולל את ההרשאה datalineage.locations.processOpenLineageMessage, אז אם לחשבון השירות בהתאמה אישית כבר יש את התפקיד הזה, לא צריך להקצות לו תפקיד נוסף של שושלת נתונים.
‫Data Lineage Events Producer (roles/datalineage.producer): התפקיד העצמאי המינימלי של שושלת נתונים שנדרש לחשבון שירות בהתאמה אישית כדי לפרסם אירועי שושלת. צריך להקצות את התפקיד הזה אם לחשבון השירות אין את התפקיד Dataproc Worker.
עריכת שושלת נתונים (roles/datalineage.editor): הרשאות לערוך ולמחוק תהליכים ואירועים של שושלת נתונים, בנוסף לפרסום שלהם. התפקיד הזה מספק יותר הרשאות ממה שנדרש להפעלה רגילה של צינור עיבוד הנתונים, וצריך אותו רק אם עומסי העבודה צריכים לשנות מטא-נתונים קיימים של שושלת נתונים.
אדמין של שושלת נתונים (roles/datalineage.admin): הרשאת אדמין מלאה למשאבים של שושלת נתונים. התפקיד הזה לא נדרש לחשבונות שירות של עומסי עבודה.

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

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

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

אחרי שמפעילים את Spark data lineage ברמת הפרויקט, התכונה תופעל במשימות 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.dataproc.lineage.enabled לערך false כשיוצרים את הסשן האינטראקטיבי או את פרופיל זמן הריצה (תבנית הסשן).

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