"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 כדי לתרגם תוכן לשפה המועדפת עליך. בתרגומים כאלו עשויות להיות שגיאות.

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

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

הקטלוג הייעודי לזמן ריצה של Lakehouse מטפל בסנכרון המטא-נתונים ובגישה מאובטחת לנתונים באמצעות הנפקת פרטי כניסה. כש-Spark שולח שאילתה לטבלה מאוחדת, הקטלוג הייעודי לזמן ריצה של Lakehouse מתרגם את המטא-נתונים של הקטלוג ומנפיק פרטי כניסה מוגבלים לטווח קצר, כדי ש-Spark יוכל לקרוא נתונים מ-Amazon S3 או מ-Cloud Storage, ומתייחס לטבלאות כאילו הן מקומיות בסביבתGoogle Cloud .

סוגים נתמכים של קטלוגים מאוחדים

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

סוג הקטלוג המאוחד	מערכות אחסון עורפיות נתמכות	מדריך להגדרה בקטלוג ייעודי לזמן ריצה של Lakehouse
‫AWS Glue Data Catalog	‫Amazon S3 (`s3://` – קטלוג ברירת המחדל של חשבון AWS או קטגוריית טבלאות של Amazon S3)	התחברות ל-AWS Glue
Databricks Unity Catalog	מיקומים חיצוניים ב-Amazon S3 (`s3://`) או ב-Cloud Storage (`gs://`)	התחברות ל-Databricks Unity Catalog
Snowflake Horizon Catalog	‫Amazon S3‏ (`s3://`) או Cloud Storage‏ (`gs://`)	התחברות לקטלוג Snowflake Horizon

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

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

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

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

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

Verify that billing is enabled for your Google Cloud project.

Enable the Dataproc, BigLake, and BigQuery 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

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

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

Verify that billing is enabled for your Google Cloud project.

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

Roles required to enable APIs

Enable the APIs

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

התפקידים הנדרשים בניהול הזהויות והרשאות הגישה (IAM)

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

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

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

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

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

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

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

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

הרצת משימות של Managed Service for Apache Spark באשכול או בסביבת זמן ריצה ללא שרת (serverless) בגרסה 2.3 או גרסה קודמת: Dataproc Worker (roles/dataproc.worker) בפרויקט
מריצים משימות של Managed Service for Apache Spark בסביבת זמן ריצה ללא שרת (serverless) בגרסה 3.0 ואילך:
- ‫Dataproc Serverless Node ‏ (roles/dataproc.serverlessNode) בפרויקט
- סוכן שירות של צומת Dataproc Resource Manager ‏ (roles/dataprocrm.nodeServiceAgent) בפרויקט
חיפוש וקריאה של מטא-נתונים של טבלה מאוחדת בזמן הריצה: BigLake Viewer (roles/biglake.viewer) בפרויקט שמכיל את הקטלוג המאוחד

כדי לוודא שלסוכן השירות של Managed Service for Apache Spark ‏ (service-PROJECT_NUMBER@dataproc-accounts.iam.gserviceaccount.com) יש את ההרשאות הנדרשות לפתרון מאפיין ההגדרה הפשוט dataproc.lakehouse.catalog. כשיוצרים אשכול, שולחים עומס עבודה של אצווה או יוצרים סשן אינטראקטיבי, צריך לבקש מהאדמין להקצות לסוכן השירות של Managed Service for Apache Spark ‏ (service-PROJECT_NUMBER@dataproc-accounts.iam.gserviceaccount.com) את תפקיד ה-IAM ‏BigLake Viewer ‏ (roles/biglake.viewer) בפרויקט שמכיל את הקטלוג המאוחד.

חשוב: צריך להקצות את התפקיד הזה לסוכן השירות של Managed Service for Apache Spark (service-PROJECT_NUMBER@dataproc-accounts.iam.gserviceaccount.com), ולא לחשבון המשתמש שלכם. אם לא תעניקו את התפקיד לחשבון הראשי הנכון, יכול להיות שתקבלו שגיאות הרשאה.

הגדרת קישוריות רשת יוצאת ל-Amazon S3

כששולחים שאילתות לקטלוגים מאוחדים שמגובים ב-Amazon S3, עומסי עבודה של הפעלות אינטראקטיביות ועיבוד באצווה ללא שרת (serverless) ב-Managed Service for Apache Spark, ואשכולות של Managed Service for Apache Spark עם כתובת IP פנימית, לא יכולים להתחבר לנקודות קצה של Amazon S3 באמצעות הגדרות הרשת שמוגדרות כברירת מחדל. מקצים Cloud Router ו-Cloud NAT ברשת ה-VPC של עומס העבודה, כדי לאפשר תנועה יוצאת ל-Amazon S3:

gcloud compute routers create lakehouse-router \
    --network=NETWORK_NAME \
    --region=REGION

gcloud compute routers nats create lakehouse-nat \
    --router=lakehouse-router \
    --auto-allocate-nat-external-ips \
    --nat-all-subnet-ip-ranges \
    --region=REGION

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

‫NETWORK_NAME: רשת ה-VPC של האשכול, עומס העבודה באצווה או הסשן האינטראקטיבי ב-Managed Service for Apache Spark, לדוגמה default.
‫REGION: האזור של Google Cloudלאשכול, לעומס העבודה באצווה או לסשן האינטראקטיבי ב-Managed Service for Apache Spark.

הגדרת Managed Service for Apache Spark ושאילתות בטבלאות מאוחדות

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

גם בהגדרה הפשוטה וגם בהגדרה הידנית, משתמשים בהרשאה של Google ל-Iceberg (GoogleAuthManager), שדורשת אחת מהגרסאות הבאות של קובץ האימג' של האשכול או של זמן הריצה ללא שרת (serverless):

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

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

אשכול

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

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

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

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.lakehouse.catalog.CATALOG_NAME=projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME,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 המקומי, לדוגמה federated_catalog. יכול להיות זהה לערך של FEDERATED_CATALOG_NAME.
‫FEDERATED_CATALOG_NAME: המזהה של הקטלוג המאוחד שיצרתם בקטע לפני שמתחילים.

שאילתות בקטלוג Snowflake Horizon

אם אתם שולחים שאילתה ל-Snowflake Horizon Catalog, שימו לב לנקודות הבאות:

‫אחסון ב-Snowflake: הגדרה פשוטה יותר תומכת בטבלאות Snowflake שמאוחסנות ב-Amazon S3. בטבלאות Snowflake שמאוחסנות ב-Cloud Storage, משתמשים בהגדרה ידנית כש-io-impl מוגדר ל-org.apache.iceberg.gcp.gcs.GCSFileIO.
עמודות Snowflake STRING: אם נתקלתם בשגיאות בשאילתות של עמודות Snowflake STRING שבהן מדיניות הסדרות של אחסון הטבלה ב-Snowflake היא לא COMPATIBLE, אתם יכולים לשנות את מדיניות הסדרות של האחסון ל-COMPATIBLE ב-Snowflake או להשבית את הווקטוריזציה של Parquet ב-Spark על ידי הוספת המאפיינים הבאים לדגל --properties:
```
spark:spark.sql.iceberg.vectorization.enabled=false,spark:spark.sql.catalog.CATALOG_NAME.table-override.read.parquet.vectorization.enabled=false
    
```

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

from pyspark.sql import SparkSession

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

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

הגדרה ידנית

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

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', 'bl://projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME') \
.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}.header.X-Iceberg-Access-Delegation', 'vended-credentials') \
.config(f'spark.sql.catalog.{catalog_name}.io-impl', 'IO_IMPL') \
.config(f'spark.sql.catalog.{catalog_name}.s3.client-factory-impl', 'org.apache.iceberg.aws.s3.DefaultS3FileIOAwsClientFactory') \
.config('spark.sql.extensions', 'org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions') \
.getOrCreate()

מחליפים את IO_IMPL בהטמעה של FileIO לאחסון שמכיל את הטבלאות:

טבלאות ב-Amazon S3: org.apache.iceberg.aws.s3.S3FileIO
טבלאות ב-Cloud Storage: org.apache.iceberg.gcp.gcs.GCSFileIO

המאפיין s3.client-factory-impl רלוונטי רק לטבלאות ב-Amazon S3. בטבלאות של Cloud Storage, משמיטים את השורה הזו.

ללא שרת (serverless)

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

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

כששולחים את עומס העבודה באצווה, מעבירים את המאפיין של הקטלוג המאוחד בלי הקידומת 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/FEDERATED_CATALOG_NAME,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. בקטע 'דרישות' שמופיע למעלה מפורטות הגרסאות הנתמכות. התפקידים שצריך ב-IAM תלויים בגרסת זמן הריצה.
‫CATALOG_NAME: שם לקטלוג Spark המקומי, לדוגמה federated_catalog. הערך יכול להיות זהה לערך של FEDERATED_CATALOG_NAME.
‫FEDERATED_CATALOG_NAME: המזהה של הקטלוג המאוחד שיצרתם בקטע לפני שמתחילים.

שליחת שאילתות ל-Snowflake Horizon Catalog

אם אתם שולחים שאילתה ל-Snowflake Horizon Catalog, שימו לב לנקודות הבאות:

אחסון ב-Snowflake: הגדרה פשוטה תומכת בטבלאות Snowflake שמאוחסנות ב-Amazon S3. כדי להשתמש בטבלאות Snowflake שמאוחסנות ב-Cloud Storage, צריך להשתמש בהגדרה ידנית עם io-impl שמוגדר ל-org.apache.iceberg.gcp.gcs.GCSFileIO.
עמודות Snowflake STRING: אם נתקלתם בשגיאות בשאילתות של עמודות Snowflake STRING שבהן מדיניות הסדרות של אחסון הטבלה ב-Snowflake היא לא COMPATIBLE, אתם יכולים לשנות את מדיניות הסדרות של האחסון ל-COMPATIBLE ב-Snowflake או להשבית את הווקטוריזציה של Parquet ב-Spark על ידי הוספת המאפיינים הבאים לדגל --properties:
```
spark.sql.iceberg.vectorization.enabled=false,spark.sql.catalog.CATALOG_NAME.table-override.read.parquet.vectorization.enabled=false
    
```

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

from pyspark.sql import SparkSession

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

הגדרה ידנית

אם לא משתמשים במאפיין ההגדרה הפשוט, צריך לציין את מאפייני הקטלוג באמצעות הדגל --properties של הפקודה gcloud dataproc batches submit. במאמר פרמטרים של הגדרות אישיות ב-Spark תוכלו למצוא רשימה של ערכי פרמטרים של ספקי שירותי אחסון.

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=bl://projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME,\
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.header.X-Iceberg-Access-Delegation=vended-credentials,\
spark.sql.catalog.CATALOG_NAME.io-impl=IO_IMPL,\
spark.sql.catalog.CATALOG_NAME.s3.client-factory-impl=org.apache.iceberg.aws.s3.DefaultS3FileIOAwsClientFactory,\
spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions"

מחליפים את IO_IMPL בהטמעה של FileIO לאחסון שמכיל את הטבלאות:

טבלאות ב-Amazon S3: org.apache.iceberg.aws.s3.S3FileIO
טבלאות ב-Cloud Storage: org.apache.iceberg.gcp.gcs.GCSFileIO

המאפיין s3.client-factory-impl רלוונטי רק לטבלאות ב-Amazon S3. בטבלאות של Cloud Storage, צריך להשמיט את המאפיין הזה.

ללא שרת (serverless)

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

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

ב-Notebook שמותקנת בו החבילה dataproc-spark-connect, כמו Notebook של BigQuery Studio או של Vertex AI Workbench, מעבירים את המאפיין של הקטלוג המאוחד בלי הקידומת 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/FEDERATED_CATALOG_NAME') \
.config('spark.sql.defaultCatalog', catalog_name) \
.getOrCreate()

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

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

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

שליחת שאילתות ל-Snowflake Horizon Catalog

אם אתם שולחים שאילתה ל-Snowflake Horizon Catalog, שימו לב לנקודות הבאות:

‫Snowflake storage: הגדרה פשוטה שתומכת בטבלאות Snowflake שמאוחסנות ב-Amazon S3. בטבלאות Snowflake שמאוחסנות ב-Cloud Storage, משתמשים בהגדרה ידנית כש-io-impl מוגדר ל-org.apache.iceberg.gcp.gcs.GCSFileIO.
עמודות STRING ב-Snowflake: אם נתקלתם בשגיאות כששלחתם שאילתות לעמודות STRING ב-Snowflake, שבהן מדיניות הסריאליזציה של האחסון בטבלה ב-Snowflake היא לא COMPATIBLE, אתם יכולים לשנות את מדיניות הסריאליזציה של האחסון ל-COMPATIBLE ב-Snowflake או להשבית את הווקטוריזציה של Parquet ב-Spark. כדי לעשות את זה, מוסיפים את הקריאות הבאות של .config() ל-session builder:
```
.config('spark.sql.iceberg.vectorization.enabled', 'false') \
.config(f'spark.sql.catalog.{catalog_name}.table-override.read.parquet.vectorization.enabled', 'false')
    
```

הגדרה ידנית

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

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', 'bl://projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME') \
.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}.header.X-Iceberg-Access-Delegation', 'vended-credentials') \
.config(f'spark.sql.catalog.{catalog_name}.io-impl', 'IO_IMPL') \
.config(f'spark.sql.catalog.{catalog_name}.s3.client-factory-impl', 'org.apache.iceberg.aws.s3.DefaultS3FileIOAwsClientFactory') \
.config('spark.sql.extensions', 'org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions') \
.getOrCreate()

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

מחליפים את IO_IMPL בהטמעה של FileIO לאחסון שמכיל את הטבלאות:

טבלאות ב-Amazon S3: org.apache.iceberg.aws.s3.S3FileIO
טבלאות ב-Cloud Storage: org.apache.iceberg.gcp.gcs.GCSFileIO

המאפיין s3.client-factory-impl רלוונטי רק לטבלאות ב-Amazon S3. בטבלאות של Cloud Storage, משמיטים את השורה הזו.

שאילתות על טבלאות מאוחדות

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

קוראים טבלה מאוחדת באמצעות PySpark DataFrame API:
```
df = spark.table("CATALOG_NAME.NAMESPACE_NAME.TABLE_NAME")
df.show(10, truncate=False)
```
מחליפים את מה שכתוב בשדות הבאים:
- ‫CATALOG_NAME: שם הקטלוג המקומי של Spark. אם מגדירים את spark.sql.defaultCatalog לערך CATALOG_NAME, אפשר להשמיט את הקידומת הזו.
- ‫NAMESPACE_NAME: מרחב השמות (מסד הנתונים או הסכימה) בקטלוג המאוחד.
- ‫TABLE_NAME: שם הטבלה בקטלוג המאוחד.

שולחים שאילתה לטבלה המאוגדת באמצעות Spark SQL:

spark.sql("""
    SELECT *
    FROM CATALOG_NAME.NAMESPACE_NAME.TABLE_NAME
    LIMIT 10
""").show()

איחוד או מימוש של נתונים בקטלוגים מאוחדים ומקומיים

קטלוגים מאוחדים הם לקריאה בלבד. כדי לצרף נתונים מרוחקים חוצי-עננים לטבלאות מקומיות או להעביר תוצאות של שאילתות לכמה מסדי נתונים ל-Cloud Storage, צריך לצרף את הקטלוג המאוחד ואת הקטלוג לקריאה ולכתיבה בקטלוג הייעודי לזמן ריצה של Lakehouse באמצעות מאפייני dataproc.lakehouse.catalog. שמופרדים בפסיקים:

אשכול

דוגמה לאשכול Managed Service for Apache Spark:

--properties="dataproc:dataproc.lakehouse.catalog.FEDERATED_SPARK_CATALOG=projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME,dataproc:dataproc.lakehouse.catalog.LOCAL_SPARK_CATALOG=projects/PROJECT_ID/catalogs/LOCAL_CATALOG_ID,spark:spark.sql.defaultCatalog=LOCAL_SPARK_CATALOG"

ללא שרת (serverless)

דוגמה לעומס עבודה של אצווה ללא שרת ב-Managed Service for Apache Spark:

--properties="dataproc.lakehouse.catalog.FEDERATED_SPARK_CATALOG=projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME,dataproc.lakehouse.catalog.LOCAL_SPARK_CATALOG=projects/PROJECT_ID/catalogs/LOCAL_CATALOG_ID,spark.sql.defaultCatalog=LOCAL_SPARK_CATALOG"

ללא שרת (serverless)

דוגמה לסשן אינטראקטיבי ללא שרת (serverless) ב-Managed Service for Apache Spark:

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

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

spark = DataprocSparkSession.builder.appName("APP_NAME") \
  .dataprocSessionConfig(session_config) \
  .config('dataproc.lakehouse.catalog.FEDERATED_SPARK_CATALOG', 'projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME') \
  .config('dataproc.lakehouse.catalog.LOCAL_SPARK_CATALOG', 'projects/PROJECT_ID/catalogs/LOCAL_CATALOG_ID') \
  .config('spark.sql.defaultCatalog', 'LOCAL_SPARK_CATALOG') \
  .getOrCreate()

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

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

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

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

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

בדוגמה הבאה נוצרת טבלה בקטלוג המקומי מטבלה מאוחדת:

spark.sql("""
    CREATE TABLE IF NOT EXISTS LOCAL_SPARK_CATALOG.LOCAL_NAMESPACE.materialized_table
    USING iceberg
    AS SELECT *
    FROM FEDERATED_SPARK_CATALOG.REMOTE_NAMESPACE.REMOTE_TABLE
""")

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

סוגים נתמכים של קטלוגים מאוחדים

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

התפקידים הנדרשים בניהול הזהויות והרשאות הגישה (IAM)

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

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

הגדרת קישוריות רשת יוצאת ל-Amazon S3

הגדרת Managed Service for Apache Spark ושאילתות בטבלאות מאוחדות

אשכול

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

שאילתות בקטלוג Snowflake Horizon

הגדרה ידנית

ללא שרת (serverless)

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

שליחת שאילתות ל-Snowflake Horizon Catalog

הגדרה ידנית

ללא שרת (serverless)

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

שליחת שאילתות ל-Snowflake Horizon Catalog

הגדרה ידנית

שאילתות על טבלאות מאוחדות

איחוד או מימוש של נתונים בקטלוגים מאוחדים ומקומיים

אשכול

ללא שרת (serverless)

ללא שרת (serverless)

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