יצירת Lakehouse באמצעות Spark וקטלוג ייעודי לזמן ריצה של Lakehouse

ארכיטקטורת lakehouse משלבת את הגמישות של אגם נתונים (data lake) עם תכונות ניהול הנתונים של מחסן נתונים (data warehouse). במאמר הזה מוסבר איך להגדיר לייקהאוס ב- Google Cloud. אתם משתמשים ב-Apache Iceberg כפורמט הטבלה, ב-Managed Service for Apache Spark לעיבוד ובקטלוג הייעודי לזמן ריצה של Lakehouse, ‏Iceberg REST Catalog, לניהול מאוחד של מטא-נתונים.

הארכיטקטורה הזו משתמשת בפורמטים של טבלאות פתוחות כמו Iceberg כדי להוסיף יכולות של מחסן נתונים, כמו טרנזקציות ופיתוח סכימות, לנתונים ב-Cloud Storage. הגישה הזו יוצרת מקור אמת יחיד (SSOT) לנתונים, שמנועים שונים יכולים לגשת אליו.

במאמרי עזרה זה מוסבר על קטלוגים שיוצרים בקטלוג ייעודי לזמן ריצה של Lakehouse. כדי לשלוח שאילתות לטבלאות Iceberg שמנוהלות על ידי AWS Glue Data Catalog,‏ Databricks Unity Catalog או Snowflake Horizon Catalog באמצעות קטלוגים מאוחדים של קטלוג ייעודי לזמן ריצה של Lakehouse, קראו את המאמר שליחת שאילתות לקטלוגים מאוחדים באמצעות Spark.

תרשים שמציג את הרכיבים של ארכיטקטורת Lakehouse, כולל Managed Service for Apache Spark,‏ Cloud Storage ו-Lakehouse REST Catalog.
דיאגרמת ארכיטקטורה של Lakehouse

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

  1. נכנסים לחשבון Google Cloud . אנחנו ממליצים למשתמשים חדשים ב- Google Cloud ליצור חשבון כדי שיוכלו להעריך את הביצועים של המוצרים שלנו בתרחישים מהעולם האמיתי. לקוחות חדשים מקבלים בחינם גם קרדיט בשווי 300$ להרצה, לבדיקה ולפריסה של עומסי העבודה.

  2. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

    Roles required to select or create a project

    • Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
    • Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

    Go to project selector

  3. Verify that you have the permissions required to complete this guide.

  4. Verify that billing is enabled for your Google Cloud project.

  5. Enable the Dataproc, BigLake, BigQuery, and Cloud Storage APIs, if any are not already enabled.

    Roles required to enable APIs

    To enable APIs, you need the serviceusage.services.enable permission. If you created the project, then you likely already have this permission through the Owner role (roles/owner). Otherwise, you can get this permission through the Service Usage Admin role (roles/serviceusage.serviceUsageAdmin). Learn how to grant roles.

    Enable the APIs

  6. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

    Roles required to select or create a project

    • Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
    • Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

    Go to project selector

  7. Verify that you have the permissions required to complete this guide.

  8. Verify that billing is enabled for your Google Cloud project.

  9. Enable the Dataproc, BigLake, BigQuery, and Cloud Storage APIs, if any are not already enabled.

    Roles required to enable APIs

    To enable APIs, you need the serviceusage.services.enable permission. If you created the project, then you likely already have this permission through the Owner role (roles/owner). Otherwise, you can get this permission through the Service Usage Admin role (roles/serviceusage.serviceUsageAdmin). Learn how to grant roles.

    Enable the APIs

    10.
  10. יצירת קטגוריה של Cloud Storage לאחסון נתוני Iceberg.
  11. יוצרים קטלוג ייעודי לזמן ריצה ב-Lakehouse שמפנה לקטגוריה. מומלץ להשתמש בקטלוג עם כמה קטגוריות.

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

כדי להריץ את הדוגמאות שבדף הזה, צריך תפקידים מסוימים בניהול הזהויות והרשאות הגישה (IAM). בהתאם למדיניות הארגון, יכול להיות שהתפקידים האלה כבר הוקצו. כדי לבדוק את התפקידים שהוקצו, קראו את המאמר האם צריך להקצות תפקידים?.

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

התפקידים שאתם צריכים תלויים באופן שבו עומס העבודה פועל. אשכולות של Managed Service for Apache Spark וסביבות זמן ריצה ללא שרת (serverless) של Managed Service for Apache Spark בגרסה 2.3 וגרסאות קודמות מריצים עומסי עבודה כחשבון שירות. בסביבות זמן ריצה ללא שרת (serverless) של Managed Service for Apache Spark מגרסה 3.0 ואילך, עומסי העבודה פועלים כברירת מחדל עם פרטי הכניסה של משתמשי הקצה (EUC). לכן, אתם צריכים להקצות תפקידים של גישה לנתונים לחשבון המשתמש שלכם במקום לחשבון שירות. אם מציינים חשבון שירות בהתאמה אישית לסביבת זמן ריצה בגרסה 3.0 ואילך, עומס העבודה פועל עם פרטי כניסה לחשבון שירות. מידע נוסף זמין במאמר תפקידים של פריסות ללא שרת (serverless).

תפקידי משתמשים

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

  • משתמשים באשכול או בסביבת זמן ריצה ללא שרת (serverless) בגרסה 2.3 או גרסה קודמת:
  • שימוש בסביבת זמן ריצה ללא שרת (serverless) בגרסה 3.0 ואילך: עריכה ב-Dataproc Serverless (roles/dataproc.serverlessEditor) בפרויקט
  • שימוש בפרטי כניסה של חשבון שירות בסביבת זמן ריצה ללא שרת בגרסה 3.0 ואילך: משתמש בחשבון שירות (roles/iam.serviceAccountUser) בחשבון השירות המותאם אישית

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

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

  • קריאה וכתיבה של נתוני טבלה במצב של מכירת אישורים: BigLake Editor (roles/biglake.editor) בפרויקט שמכיל את הקטלוג
  • ניהול משאבי קטלוג וכתיבת נתוני טבלה במצב שאינו הפצת פרטי כניסה:
    • עריכה ב-BigLake ‏ (roles/biglake.editor) בפרויקט שמכיל את הקטלוג
    • Storage Object User (roles/storage.objectUser) בכל הקטגוריות של Cloud Storage שמשויכות לקטלוג

תפקידים של חשבון שירות

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

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

  • הרצת משימות של Managed Service for Apache Spark באשכול או בסביבת זמן ריצה ללא שרתים בגרסה 2.3 ואילך: ‫Dataproc Worker (roles/dataproc.worker) בפרויקט
  • הפעלת משימות של Managed Service for Apache Spark בסביבת זמן ריצה 3.0 ואילך ללא שרת:
  • קריאה וכתיבה של נתוני טבלה במצב של מכירת אישורים: BigLake Editor (roles/biglake.editor) בפרויקט שמכיל את הקטלוג
  • ניהול משאבי קטלוג וכתיבת נתוני טבלה במצב ללא הפצת פרטי כניסה:
    • BigLake Editor (roles/biglake.editor) בפרויקט שמכיל את הקטלוג
    • Storage Object User (roles/storage.objectUser) בכל הקטגוריות של Cloud Storage שמשויכות לקטלוג

הגדרת Managed Service for Apache Spark ויצירת סשן Spark

הגדרת אשכול של Managed Service for Apache Spark, עומס עבודה של אצווה ללא שרת (serverless) של Managed Service for Apache Spark או סשן אינטראקטיבי ללא שרת (serverless) של Managed Service for Apache Spark לשימוש בקטלוג REST של Iceberg, קטלוג ייעודי לזמן ריצה של Lakehouse. כל שלושתם תומכים במאפיין ההגדרה dataproc.lakehouse.catalog.CATALOG_NAME הפשוט (מומלץ), שמאפשר ל-Managed Service for Apache Spark לקרוא את הגדרת הקטלוג וליצור את מאפייני Spark של קטלוג Iceberg REST. רק לטופס של האשכול יש קידומת dataproc:.

גרסאות נתמכות של קובצי אימג' וסביבות זמן ריצה:

גם בהגדרה פשוטה וגם בהגדרה ידנית נעשה שימוש בהרשאה של Google ל-Iceberg ‏ (GoogleAuthManager), שנדרש בה אחת מהגרסאות הבאות:

  • אשכולות של Managed Service for Apache Spark: גרסאות אימג' של 2.2 2.2.65 ואילך, או גרסאות אימג' של 2.3 2.3.11 ואילך.
  • עומסי עבודה של אצווה ללא שרת וסשנים אינטראקטיביים ב-Managed Service for Apache Spark: סביבות זמן ריצה 2.2 2.2.60 ואילך, סביבות זמן ריצה 2.3 2.3.10 ואילך או סביבות זמן ריצה 3.0.

עבור עומסי עבודה (workloads) באצווה ללא שרת (serverless) וסשנים אינטראקטיביים, התפקידים הנדרשים תלויים בגרסת זמן הריצה. בזמני ריצה מגרסה 3.0 ואילך נעשה שימוש כברירת מחדל בהרשאות של משתמשי קצה. למידע נוסף, קראו על התפקידים הנדרשים.

אשכול

יוצרים אשכול של Managed Service for Apache Spark עם הרכיבים האופציונליים Iceberg ו-Jupyter, באמצעות מאפייני הגדרה פשוטים (מומלץ) או על ידי ציון ידני של מאפייני הקטלוג ב-Notebook.

הגדרה פשוטה באמצעות מאפיינים (מומלץ)

יוצרים את האשכול עם מאפיין הקטלוג:

gcloud dataproc clusters create CLUSTER_NAME \
  --project=PROJECT_ID \
  --region=REGION \
  --image-version=IMAGE_VERSION \
  --optional-components=ICEBERG,JUPYTER \
  --enable-component-gateway \
  --properties='dataproc:dataproc.lineage.enabled=true,dataproc:dataproc.lakehouse.catalog.CATALOG_NAME=projects/PROJECT_ID/catalogs/CATALOG_ID,spark:spark.sql.defaultCatalog=CATALOG_NAME'

מחליפים את מה שכתוב בשדות הבאים:

  • ‫CLUSTER_NAME: שם לאשכול.
  • ‫PROJECT_ID: מזהה הפרויקט ב-Google Cloud .
  • ‫REGION: האזור של Google Cloudלאשכול, לדוגמה us-central1.
  • ‫IMAGE_VERSION: גרסת האימג' של Managed Service for Apache Spark, לדוגמה 2.3-debian12. גרסאות נתמכות של קובצי אימג' וסביבות זמן ריצה
  • ‫CATALOG_NAME: שם לקטלוג Spark המקומי, לדוגמה iceberg_catalog. הוא יכול להיות זהה לערך של CATALOG_ID.
  • ‫CATALOG_ID: המזהה של הקטלוג שיצרתם בקטע לפני שמתחילים.

המאפיין dataproc:dataproc.lineage.enabled=true לא נדרש כדי שקטלוג ה-REST של Iceberg בקטלוג הייעודי לזמן ריצה של Lakehouse יפעל כמו שצריך. הוא נוסף למעקב אחר שושלת הנתונים בדוגמה לשושלת נתונים.

מתחברים לאשכול באמצעות notebook של Jupyter (‏notebook של Vertex AI Workbench או notebook שהופעל באשכול), ויוצרים את SparkSession בלי לציין הגדרות קטלוג:

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("APP_NAME").getOrCreate()

מחליפים את APP_NAME בשם של אפליקציית Spark.

הגדרה ידנית

אם אתם לא משתמשים במאפיין ההגדרה הפשוט, צרו את האשכול כמו שמתואר למעלה בלי dataproc:dataproc.lakehouse.catalog. בדגל --properties, התחברו ל-Jupyter Notebook והגדירו את SparkSession באופן ידני, באופן הבא:

import pyspark
from pyspark.context import SparkContext
from pyspark.sql import SparkSession

catalog_name = "CATALOG_NAME"

spark = SparkSession.builder.appName("APP_NAME") \
.config('spark.sql.defaultCatalog', catalog_name) \
.config(f'spark.sql.catalog.{catalog_name}', 'org.apache.iceberg.spark.SparkCatalog') \
.config(f'spark.sql.catalog.{catalog_name}.type', 'rest') \
.config(f'spark.sql.catalog.{catalog_name}.uri', 'https://biglake.googleapis.com/iceberg/v1/restcatalog') \
.config(f'spark.sql.catalog.{catalog_name}.warehouse', 'WAREHOUSE_PATH') \
.config(f'spark.sql.catalog.{catalog_name}.header.x-goog-user-project', 'PROJECT_ID') \
.config(f'spark.sql.catalog.{catalog_name}.rest.auth.type', 'org.apache.iceberg.gcp.auth.GoogleAuthManager') \
.config(f'spark.sql.catalog.{catalog_name}.io-impl', 'org.apache.iceberg.gcp.gcs.GCSFileIO') \
.config('spark.sql.extensions', 'org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions') \
.getOrCreate()

מחליפים את מה שכתוב בשדות הבאים:

  • ‫CATALOG_NAME: שם לקטלוג המקומי של Spark, לדוגמה, iceberg_catalog.
  • ‫APP_NAME: השם של אפליקציית Spark.
  • ‫WAREHOUSE_PATH: הנתיב למחסן הנתונים. בקטלוגים עם כמה קטגוריות (מומלץ), משתמשים ב-bl://projects/PROJECT_ID/catalogs/CATALOG_ID. לספריות עם דלי אחד, משתמשים ב-gs://CLOUD_STORAGE_BUCKET_NAME.
  • ‫PROJECT_ID: מזהה הפרויקט ב- Google Cloud .
הגדרה באמצעות הקצאת הרשאות

כדי להשתמש במכירת פרטי כניסה, צריך להשתמש בקטלוג במצב מכירת פרטי כניסה ולהוסיף את הכותרת X-Iceberg-Access-Delegation לבקשות קטלוג REST של Iceberg עם הערך vended-credentials. לשם כך, מוסיפים את השורה הבאה לבונה SparkSession:

.config(f'spark.sql.catalog.{catalog_name}.header.X-Iceberg-Access-Delegation', 'vended-credentials')

בדוגמה הבאה מוגדר SparkSession עם מכירת אישורים:

from pyspark.sql import SparkSession

catalog_name = "CATALOG_NAME"

spark = SparkSession.builder.appName("APP_NAME") \
.config('spark.sql.defaultCatalog', catalog_name) \
.config(f'spark.sql.catalog.{catalog_name}', 'org.apache.iceberg.spark.SparkCatalog') \
.config(f'spark.sql.catalog.{catalog_name}.type', 'rest') \
.config(f'spark.sql.catalog.{catalog_name}.uri', 'https://biglake.googleapis.com/iceberg/v1/restcatalog') \
.config(f'spark.sql.catalog.{catalog_name}.warehouse', 'WAREHOUSE_PATH') \
.config(f'spark.sql.catalog.{catalog_name}.header.x-goog-user-project', 'PROJECT_ID') \
.config(f'spark.sql.catalog.{catalog_name}.rest.auth.type', 'org.apache.iceberg.gcp.auth.GoogleAuthManager') \
.config(f'spark.sql.catalog.{catalog_name}.io-impl', 'org.apache.iceberg.gcp.gcs.GCSFileIO') \
.config(f'spark.sql.catalog.{catalog_name}.header.X-Iceberg-Access-Delegation', 'vended-credentials') \
.config('spark.sql.extensions', 'org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions') \
.getOrCreate()

ההגדרה הפשוטה מוסיפה את הכותרת הזו כשהקטלוג משתמש במצב של מכירת אישורים. מידע נוסף זמין בקטע Headers in the RESTCatalog במסמכי התיעוד של Iceberg.

ללא שרת (serverless)

שליחת עומס עבודה (workload) של PySpark batch אל Managed Service for Apache Spark serverless באמצעות מאפייני הגדרה פשוטים (מומלץ), או על ידי ציון ידני של מאפייני קטלוג.

הגדרה פשוטה באמצעות מאפיינים (מומלץ)

מעבירים את הנכס dataproc.lakehouse.catalog.CATALOG_NAME (ללא הקידומת dataproc:) כששולחים את עומס העבודה של האצווה:

gcloud dataproc batches submit pyspark PYSPARK_FILE \
  --project=PROJECT_ID \
  --region=REGION \
  --version=RUNTIME_VERSION \
  --properties="dataproc.lakehouse.catalog.CATALOG_NAME=projects/PROJECT_ID/catalogs/CATALOG_ID,spark.sql.defaultCatalog=CATALOG_NAME"

מחליפים את מה שכתוב בשדות הבאים:

  • ‫PYSPARK_FILE: gs:// הנתיב ב-Cloud Storage לקובץ של אפליקציית PySpark.
  • ‫PROJECT_ID: מזהה הפרויקט ב-Google Cloud .
  • ‫REGION: האזור של עומס העבודה באצווה.
  • ‫RUNTIME_VERSION: גרסת סביבת זמן הריצה ללא שרת (serverless) של Managed Service for Apache Spark, לדוגמה 3.0. התפקידים הנדרשים תלויים בגרסת זמן הריצה (ראו גרסאות נתמכות של קובצי אימג' וזמן ריצה).
  • ‫CATALOG_NAME: שם לקטלוג Spark המקומי, לדוגמה iceberg_catalog. הוא יכול להיות זהה לערך של CATALOG_ID.
  • ‫CATALOG_ID: המזהה של הקטלוג שיצרתם בקטע לפני שמתחילים.

בקובץ האפליקציה של PySpark, יוצרים את SparkSession בלי לציין את הגדרות הקטלוג:

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("APP_NAME").getOrCreate()

הגדרה ידנית

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

gcloud dataproc batches submit pyspark PYSPARK_FILE \
  --project=PROJECT_ID \
  --region=REGION \
  --version=RUNTIME_VERSION \
  --properties="\
spark.sql.defaultCatalog=CATALOG_NAME,\
spark.sql.catalog.CATALOG_NAME=org.apache.iceberg.spark.SparkCatalog,\
spark.sql.catalog.CATALOG_NAME.type=rest,\
spark.sql.catalog.CATALOG_NAME.uri=https://biglake.googleapis.com/iceberg/v1/restcatalog,\
spark.sql.catalog.CATALOG_NAME.warehouse=WAREHOUSE_PATH,\
spark.sql.catalog.CATALOG_NAME.header.x-goog-user-project=PROJECT_ID,\
spark.sql.catalog.CATALOG_NAME.rest.auth.type=org.apache.iceberg.gcp.auth.GoogleAuthManager,\
spark.sql.catalog.CATALOG_NAME.io-impl=org.apache.iceberg.gcp.gcs.GCSFileIO,\
spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions"

מחליפים את WAREHOUSE_PATH בנתיב למחסן הנתונים. בקטלוגים עם כמה דליים (מומלץ), משתמשים ב-bl://projects/PROJECT_ID/catalogs/CATALOG_ID. לקטלוגים עם קטגוריה אחת, משתמשים ב-gs://CLOUD_STORAGE_BUCKET_NAME.

הגדרה באמצעות הקצאת הרשאות

כדי להשתמש במכירת אישורים, צריך להשתמש בקטלוג במצב מכירת אישורים ולהוסיף את הכותרת X-Iceberg-Access-Delegation לבקשות קטלוג של Iceberg REST עם הערך vended-credentials על ידי הוספת המאפיין הבא לדגל --properties:

spark.sql.catalog.CATALOG_NAME.header.X-Iceberg-Access-Delegation=vended-credentials

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

gcloud dataproc batches submit pyspark PYSPARK_FILE \
  --project=PROJECT_ID \
  --region=REGION \
  --version=RUNTIME_VERSION \
  --properties="\
spark.sql.defaultCatalog=CATALOG_NAME,\
spark.sql.catalog.CATALOG_NAME=org.apache.iceberg.spark.SparkCatalog,\
spark.sql.catalog.CATALOG_NAME.type=rest,\
spark.sql.catalog.CATALOG_NAME.uri=https://biglake.googleapis.com/iceberg/v1/restcatalog,\
spark.sql.catalog.CATALOG_NAME.warehouse=WAREHOUSE_PATH,\
spark.sql.catalog.CATALOG_NAME.header.x-goog-user-project=PROJECT_ID,\
spark.sql.catalog.CATALOG_NAME.rest.auth.type=org.apache.iceberg.gcp.auth.GoogleAuthManager,\
spark.sql.catalog.CATALOG_NAME.io-impl=org.apache.iceberg.gcp.gcs.GCSFileIO,\
spark.sql.catalog.CATALOG_NAME.header.X-Iceberg-Access-Delegation=vended-credentials,\
spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions"

ההגדרה הפשוטה מוסיפה את הכותרת הזו כשהקטלוג משתמש במצב של מכירת אישורים. מידע נוסף זמין בקטע Headers in RESTCatalog במאמרי העזרה של Iceberg.

ללא שרת (serverless)

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

הגדרה פשוטה באמצעות מאפיינים (מומלץ)

ב-Notebook שמותקנת בו החבילה dataproc-spark-connect, כמו Notebook של BigQuery Studio או Notebook של Vertex AI Workbench, מעבירים את המאפיין dataproc.lakehouse.catalog.CATALOG_NAME (בלי הקידומת dataproc:) כשיוצרים את הסשן:

from google.cloud.dataproc_spark_connect import DataprocSparkSession
from google.cloud.dataproc_v1 import Session

catalog_name = "CATALOG_NAME"

session_config = Session()
session_config.runtime_config.version = 'RUNTIME_VERSION'

spark = DataprocSparkSession.builder.appName("APP_NAME") \
.dataprocSessionConfig(session_config) \
.config(f'dataproc.lakehouse.catalog.{catalog_name}', 'projects/PROJECT_ID/catalogs/CATALOG_ID') \
.config('spark.sql.defaultCatalog', catalog_name) \
.getOrCreate()

מחליפים את מה שכתוב בשדות הבאים:

  • ‫CATALOG_NAME: שם לקטלוג המקומי של Spark, לדוגמה, iceberg_catalog. הוא יכול להיות זהה לערך של CATALOG_ID.
  • ‫RUNTIME_VERSION: גרסת סביבת זמן הריצה ללא שרת (serverless) של Managed Service for Apache Spark, לדוגמה 3.0. התפקידים הנדרשים תלויים בגרסת זמן הריצה (ראו גרסאות נתמכות של תמונות וזמן ריצה).
  • ‫APP_NAME: השם של אפליקציית Spark.
  • ‫PROJECT_ID: מזהה הפרויקט ב-Google Cloud .
  • ‫CATALOG_ID: המזהה של הקטלוג שיצרתם בקטע לפני שמתחילים.

אפשר גם להגדיר את המאפיינים האלה בתבנית סשן.

הגדרה ידנית

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

from google.cloud.dataproc_spark_connect import DataprocSparkSession
from google.cloud.dataproc_v1 import Session

catalog_name = "CATALOG_NAME"

session_config = Session()
session_config.runtime_config.version = 'RUNTIME_VERSION'

spark = DataprocSparkSession.builder.appName("APP_NAME") \
.dataprocSessionConfig(session_config) \
.config('spark.sql.defaultCatalog', catalog_name) \
.config(f'spark.sql.catalog.{catalog_name}', 'org.apache.iceberg.spark.SparkCatalog') \
.config(f'spark.sql.catalog.{catalog_name}.type', 'rest') \
.config(f'spark.sql.catalog.{catalog_name}.uri', 'https://biglake.googleapis.com/iceberg/v1/restcatalog') \
.config(f'spark.sql.catalog.{catalog_name}.warehouse', 'WAREHOUSE_PATH') \
.config(f'spark.sql.catalog.{catalog_name}.header.x-goog-user-project', 'PROJECT_ID') \
.config(f'spark.sql.catalog.{catalog_name}.rest.auth.type', 'org.apache.iceberg.gcp.auth.GoogleAuthManager') \
.config(f'spark.sql.catalog.{catalog_name}.io-impl', 'org.apache.iceberg.gcp.gcs.GCSFileIO') \
.config('spark.sql.extensions', 'org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions') \
.getOrCreate()

מחליפים את RUNTIME_VERSION כמו שמתואר בדוגמה הפשוטה להגדרה שמופיעה בהמשך הכרטיסייה הזו.

מחליפים את WAREHOUSE_PATH בנתיב למחסן הנתונים. בקטלוגים עם כמה קטגוריות (מומלץ), משתמשים ב-bl://projects/PROJECT_ID/catalogs/CATALOG_ID. לקטלוגים עם קטגוריה אחת, משתמשים ב-gs://CLOUD_STORAGE_BUCKET_NAME.

הגדרה באמצעות הקצאת הרשאות

כדי להשתמש בהנפקת פרטי כניסה, צריך להשתמש בקטלוג במצב הנפקת פרטי כניסה ולהוסיף את הכותרת X-Iceberg-Access-Delegation לבקשות של קטלוג Iceberg REST עם הערך vended-credentials, על ידי הוספת השורה הבאה ל-session builder:

.config(f'spark.sql.catalog.{catalog_name}.header.X-Iceberg-Access-Delegation', 'vended-credentials')

בדוגמה הבאה נוצר סשן עם הפצת פרטי כניסה:

from google.cloud.dataproc_spark_connect import DataprocSparkSession
from google.cloud.dataproc_v1 import Session

catalog_name = "CATALOG_NAME"

session_config = Session()
session_config.runtime_config.version = 'RUNTIME_VERSION'

spark = DataprocSparkSession.builder.appName("APP_NAME") \
.dataprocSessionConfig(session_config) \
.config('spark.sql.defaultCatalog', catalog_name) \
.config(f'spark.sql.catalog.{catalog_name}', 'org.apache.iceberg.spark.SparkCatalog') \
.config(f'spark.sql.catalog.{catalog_name}.type', 'rest') \
.config(f'spark.sql.catalog.{catalog_name}.uri', 'https://biglake.googleapis.com/iceberg/v1/restcatalog') \
.config(f'spark.sql.catalog.{catalog_name}.warehouse', 'WAREHOUSE_PATH') \
.config(f'spark.sql.catalog.{catalog_name}.header.x-goog-user-project', 'PROJECT_ID') \
.config(f'spark.sql.catalog.{catalog_name}.rest.auth.type', 'org.apache.iceberg.gcp.auth.GoogleAuthManager') \
.config(f'spark.sql.catalog.{catalog_name}.io-impl', 'org.apache.iceberg.gcp.gcs.GCSFileIO') \
.config(f'spark.sql.catalog.{catalog_name}.header.X-Iceberg-Access-Delegation', 'vended-credentials') \
.config('spark.sql.extensions', 'org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions') \
.getOrCreate()

ההגדרה הפשוטה מוסיפה את הכותרת הזו כשהקטלוג משתמש במצב של הפצת פרטי כניסה. מידע נוסף זמין בקטע Headers in the RESTCatalog במסמכי התיעוד של Iceberg.

ניהול נתונים באמצעות Spark SQL

אחרי שמגדירים את הסשן של Spark, משתמשים ב-Spark SQL כדי לבצע פעולות של ניהול נתונים.

הגדרת הקטלוג: המאפיין dataproc.lakehouse.catalog. רושם את הקטלוג ב-Spark, אבל הוא לא הופך אותו לקטלוג ברירת המחדל. בדוגמאות שבקטע הקודם מוגדר spark.sql.defaultCatalog (spark:spark.sql.defaultCatalog באשכול) כדי שאפשר יהיה להשתמש בשמות טבלאות לא מלאים. אם משמיטים את המאפיין הזה, צריך להוסיף לשמות הטבלאות את שם הקטלוג, לדוגמה: CATALOG_NAME.NAMESPACE_NAME.base_table.

  1. יוצרים מרחב שמות. בקטלוג של זמן הריצה של Lakehouse, Iceberg REST Catalog, מרחב שמות תואם למערך נתונים ב-BigQuery.

    spark.sql("CREATE NAMESPACE IF NOT EXISTS NAMESPACE_NAME")
spark.sql("USE NAMESPACE_NAME")

    מחליפים את NAMESPACE_NAME בשם של מרחב השמות, לדוגמה, spark_lakehouse.

  2. יוצרים מסד נתונים טבלאי בפורמט Iceberg ומזינים נתונים.

    spark.sql("DROP TABLE IF EXISTS base_table PURGE")
spark.sql("CREATE TABLE base_table (id LONG) USING iceberg")
spark.sql("INSERT INTO base_table VALUES 0, 1, 2, 3, 4")
spark.sql("SELECT * FROM base_table").show()

    הפלט אמור להיראות כך:

       +---+
   | id|
   +---+
   |  0|
   |  1|
   |  2|
   |  3|
   |  4|
   +---+

  3. ליצור טבלה אחרת לנתונים חדשים.

    spark.sql("DROP TABLE IF EXISTS newdata PURGE")
spark.sql("CREATE TABLE newdata(id LONG) USING iceberg")
spark.sql("INSERT INTO newdata VALUES 3, 4, 5, 6")
spark.sql("SELECT * FROM newdata").show()

    הפלט אמור להיראות כך:

       +---+
   | id|
   +---+
   |  3|
   |  4|
   |  5|
   |  6|
   +---+

  4. ממזגים את הנתונים החדשים במסד הנתונים הטבלאי.

    spark.sql("""MERGE INTO base_table USING newdata
           ON base_table.id = newdata.id
           WHEN MATCHED THEN
             UPDATE SET base_table.id = newdata.id
           WHEN NOT MATCHED THEN INSERT * """)
spark.sql("SELECT * FROM base_table").show()

    הפלט אמור להיראות כך:

       +---+
   | id|
   +---+
   |  0|
   |  1|
   |  2|
   |  3|
   |  4|
   |  5|
   |  6|
   +---+

  5. מעדכנים רשומות במסד הנתונים הטבלאי.

    spark.sql(
     "UPDATE base_table SET id = (id + 100) WHERE (id % 2 == 0)"
 )
spark.sql("SELECT * FROM base_table").show()

    הפלט אמור להיראות כך:

       +---+
   | id|
   +---+
   |  3|
   |104|
   |  5|
   |106|
   |100|
   |102|
   |  1|
   +---+

  6. מחיקת רשומות ממסד נתונים טבלאי.

    spark.sql("DELETE FROM base_table WHERE (id % 2 == 0)")
spark.sql("SELECT * FROM base_table").show()

    הפלט אמור להיראות כך:

       +---+
   | id|
   +---+
   |  3|
   |  5|
   |  1|
   +---+

שליחת שאילתה לגבי תמונת מצב היסטורית

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

  1. מאחזרים את מזהה התמונה של גרסת הטבלה לפני הפעולות MERGE, UPDATE ו-DELETE.

    snapshot_ids = spark.sql(
     "SELECT snapshot_id FROM `NAMESPACE_NAME`.`base_table`.snapshots"
).collect()
oldest_snapshot_id = snapshot_ids[1]["snapshot_id"]

    מחליפים את NAMESPACE_NAME במרחב השמות שיצרתם.

  2. מריצים שאילתה על הטבלה באמצעות מזהה קובץ ה-snapshot שאוחזר.

    df = (
     spark.read.format("iceberg")
     .option("versionAsOf", oldest_snapshot_id)
     .load("base_table")
 )
df.show()

    הפלט מציג את מצב הטבלה אחרי הפעולה MERGE אבל לפני הפעולות UPDATE או DELETE.

       +---+
   | id|
   +---+
   |  0|
   |  1|
   |  2|
   |  3|
   |  4|
   |  5|
   |  6|
   +---+

איך בודקים את שושלת הנתונים

אם משתמשים בגרסאות תמונה של Managed Service for Apache Spark 2.2 ומעלה, אפשר לעקוב אחרי תנועת הנתונים בין טבלאות של קטלוג ייעודי לזמן ריצה של Lakehouse לבין טבלאות של קטלוג REST של Iceberg באמצעות השתלשלות נתונים.

דוגמה לשושלת נתונים

  1. יוצרים טבלאות מקור ויעד ב-Iceberg, ואז מעתיקים את הנתונים.

    spark.sql("DROP TABLE IF EXISTS source_table PURGE")
spark.sql("DROP TABLE IF EXISTS target_table PURGE")
spark.sql("CREATE TABLE source_table (id LONG) USING iceberg")
spark.sql("""CREATE TABLE target_table
  USING ICEBERG
  AS SELECT max(id) as top_id FROM source_table
  """)

  2. נכנסים לדף Knowledge Catalog Search במסוף Google Cloud .

    לחיפוש Google

  3. מחפשים את אחת מהטבלאות ולוחצים על הכרטיסייה Lineage:

    דוגמה לתיעוד מקורות הנתונים בדף Knowledge Catalog במסוף Google Cloud .
    דוגמה לגרף של השתלשלות הנתונים בדף Knowledge Catalog במסוף Google Cloud .

    במעקב אחר שושלת הנתונים מזוהים גם הייצוגים הלוגיים (טבלת קטלוג ייעודי לזמן ריצה של Lakehouse) וגם הייצוגים הפיזיים (Cloud Storage) של טבלאות קטלוג REST של Iceberg בקטלוג ייעודי לזמן ריצה של Lakehouse.

בעיות מוכרות בשושלת הנתונים

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

המאמרים הבאים