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

יצירת טבלת Apache Iceberg עם מטא-נתונים בקטלוג של Lakehouse runtime

במסמך הזה מוסבר איך להריץ עומסי עבודה של SQL ו-PySpark ב-Managed Service for Apache Spark כדי ליצור טבלת Apache Iceberg עם מטא-נתונים שמאוחסנים בקטלוג של Lakehouse runtime. מידע על דרכים אחרות להריץ קוד Spark זמין במאמרים הפעלת קוד PySpark ב-Notebook של BigQuery והפעלת עומס עבודה של Apache Spark.

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

אם עדיין לא עשיתם זאת, יוצרים Google Cloud פרויקט וקטגוריה ב-Cloud Storage.

הגדרת הפרויקט
יוצרים קטגוריה של Cloud Storage בפרויקט.
1. במסוף Google Cloud , נכנסים לדף Buckets של Cloud Storage.
  כניסה לדף Buckets
2. לוחצים על יצירה.
3. ממלאים את פרטי הקטגוריה בדף Create a bucket. כדי לעבור לשלב הבא לוחצים על Continue.
  1. בקטע Get started (תחילת העבודה), מבצעים את הפעולות הבאות:
    - מזינים שם ייחודי בהיקף גלובלי שעומד בקריטריונים לשמות של קטגוריות.
    - כדי להוסיף תווית של דלי, מרחיבים את הקטע Labels (תוויות) (), לוחצים על Add label (הוספת תווית) ומציינים key ו-value בשביל התווית.
  2. בקטע Choose where to store your data, מבצעים את הפעולות הבאות:
    1. בוחרים סוג מיקום.
    2. בתפריט הנפתח Location type, בוחרים מיקום שבו יישמרו נתוני הקטגוריה באופן קבוע.
      - אם בוחרים את סוג המיקום בשני אזורים, אפשר גם להפעיל רפליקציה בקצב טורבו באמצעות תיבת הסימון הרלוונטית.
    3. כדי להגדיר שכפול בין מאגרי מידע, בוחרים באפשרות הוספת שכפול בין מאגרי מידע באמצעות Storage Transfer Service ופועלים לפי השלבים הבאים:
      הגדרה של רפליקציה בין מאגרי מידע
      
      בתפריט Bucket, בוחרים באפשרות הרצויה.
      
      בקטע הגדרות השכפול, לוחצים על הגדרה כדי להגדיר את ההגדרות של משימת השכפול.
      
      מופיעה החלונית Configure cross-bucket replication.
      
      כדי לסנן אובייקטים לשכפול לפי קידומת של שם האובייקט, מזינים קידומת שרוצים לכלול או להחריג אובייקטים ממנה, ואז לוחצים על הוספת קידומת.
      
      כדי להגדיר סוג אחסון לאובייקטים המשוכפלים, בוחרים סוג אחסון בתפריט סוג אחסון. אם מדלגים על השלב הזה, האובייקטים המשוכפלים ישתמשו בסוג האחסון של קטגוריית היעד כברירת מחדל.
      
      לוחצים על סיום.
  3. בקטע Choose how to store your data, מבצעים את הפעולות הבאות:
    1. בוחרים default storage class לקטגוריה או Autoclass לניהול אוטומטי של סוג האחסון (storage class) של נתוני הקטגוריה.
    2. כדי להפעיל מרחב שמות היררכי, בקטע Optimize storage for data-intensive workloads, בוחרים באפשרות Enable hierarchical namespace on this bucket.
      הערה: אי אפשר להפעיל מרחב שמות היררכי בקטגוריות קיימות.
  4. בקטע Choose how to control access to objects, בוחרים אם הקטגוריה אוכפת public access prevention או לא, ואז בוחרים שיטת בקרת גישה לאובייקטים של הקטגוריה.
    הערה: אי אפשר לשנות את ההגדרה Prevent public access אם ההגדרה הזו נאכפת על ידי מדיניות הארגון.
  5. בקטע Choose how to protect object data, מבצעים את הפעולות הבאות:
    - בוחרים באחת מהאפשרויות בקטע הגנה על נתונים שרוצים להגדיר לקטגוריה.
      - כדי להפעיל מחיקה עם אפשרות שחזור, מסמנים את התיבה Soft delete policy (For data recovery) ומציינים את מספר הימים שבהם רוצים לשמור אובייקטים אחרי המחיקה.
      - כדי להגדיר ניהול גרסאות של אובייקטים, מסמנים את התיבה Object versioning (For version control) (ניהול גרסאות של אובייקטים (לניהול גרסאות)), ומציינים את מספר הגרסאות המקסימלי לכל אובייקט ואת מספר הימים שאחריהם הגרסאות הלא עדכניות יפוגו.
      - כדי להפעיל את מדיניות שמירת הנתונים על אובייקטים וקטגוריות, לוחצים על תיבת הסימון שמירת נתונים (לצורך תאימות), ואז מבצעים את הפעולות הבאות:
        
        כדי להפעיל את הנעילה של שמירת אובייקטים, מסמנים את התיבה הפעלת שמירת אובייקטים.
        
        כדי להפעיל את נעילת הקטגוריה, מסמנים את תיבת הסימון הגדרת מדיניות שמירת נתונים בקטגוריה ובוחרים יחידת זמן ואת משך הזמן של תקופת השמירה.
    - כדי לבחור איך להצפין את נתוני האובייקט, מרחיבים את הקטע Data encryption () ובוחרים Data encryption method.
4. לוחצים על יצירה.
נותנים את התפקיד BigQuery Data Editor (roles/bigquery.dataEditor) לחשבון שירות ברירת המחדל של Compute Engine,‏ PROJECT_NUMBER-compute@developer.gserviceaccount.com. הוראות מפורטות מופיעות במאמר הקצאת תפקיד יחיד.

דוגמה ל-Google Cloud CLI:
```
gcloud projects add-iam-policy-binding PROJECT_ID \
    --member PROJECT_NUMBER-compute@developer.gserviceaccount.com \
--role roles/bigquery.dataEditor
```
הערות:
- ‫PROJECT_ID ו-PROJECT_NUMBER מופיעים בקטע Project info במרכז הבקרה של מסוף Google Cloud .

מיפוי משאבי OSS למשאבי BigQuery

שימו לב למיפוי הבא בין מונחים של משאבים בקוד פתוח לבין מונחים של משאבים ב-BigQuery:

מקור מידע על OSS	משאב BigQuery
מרחב שמות, מסד נתונים	קבוצת הנתונים
טבלה מחולקת למחיצות או לא מחולקת למחיצות	טבלה
הצגה	הצגה

יצירת טבלת Iceberg

בקטע הזה נסביר איך ליצור טבלת Iceberg עם מטא-נתונים בקטלוג של Lakehouse runtime באמצעות Managed Service for Apache Spark Spark SQL ועומסי עבודה של PySpark batch.

Spark SQL

הפעלת עומס עבודה של Spark SQL כדי ליצור טבלת Iceberg

בשלבים הבאים מוסבר איך להריץ עומס עבודה של אצווה ב-Managed Service for Apache Spark SQL כדי ליצור טבלת Iceberg עם מטא-נתונים של הטבלה שמאוחסנים בקטלוג של זמן הריצה של Lakehouse.

מעתיקים את פקודות ה-SQL הבאות של Spark באופן מקומי או ב-Cloud Shell לקובץ iceberg-table.sql.

USE CATALOG_NAME;
CREATE NAMESPACE IF NOT EXISTS example_namespace;
USE example_namespace;
DROP TABLE IF EXISTS example_table;
CREATE TABLE example_table (id int, data string) USING ICEBERG LOCATION 'gs://BUCKET/WAREHOUSE_FOLDER';
INSERT INTO example_table VALUES (1, 'first row');
ALTER TABLE example_table ADD COLUMNS (newDoubleCol double);
DESCRIBE TABLE example_table;

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

‫CATALOG_NAME: שם קטלוג Iceberg.
‫BUCKET ו-WAREHOUSE_FOLDER: קטגוריה של Cloud Storage ותיקייה שמשמשות כספריית מחסן נתונים של Iceberg.

כדי לשלוח את עומס העבודה של Spark SQL, מריצים את הפקודה הבאה באופן מקומי או ב-Cloud Shell מהספרייה שמכילה את iceberg-table.sql.
```
gcloud dataproc batches submit spark-sql iceberg-table.sql \
    --project=PROJECT_ID \
    --region=REGION \
    --deps-bucket=BUCKET_NAME \
    --version=2.2 \
    --subnet=SUBNET_NAME \
    --properties="spark.sql.catalog.CATALOG_NAME=org.apache.iceberg.spark.SparkCatalog,spark.sql.catalog.CATALOG_NAME.catalog-impl=org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog,spark.sql.catalog.CATALOG_NAME.gcp_project=PROJECT_ID,spark.sql.catalog.CATALOG_NAME.gcp_location=LOCATION,spark.sql.catalog.CATALOG_NAME.warehouse=gs://BUCKET/WAREHOUSE_FOLDER"
  
```
הערות:
- ‫PROJECT_ID: מזהה הפרויקט ב- Google Cloud . מזהי הפרויקטים מופיעים בקטע Project info בלוח הבקרה של מסוף Google Cloud .
- ‫REGION: אזור זמין ב-Compute Engine להרצת עומס העבודה.
- ‫BUCKET_NAME: שם הקטגוריה של Cloud Storage. מערכת Spark מעלה את התלויות של עומס העבודה לתיקייה /dependencies בקטגוריה הזו לפני הפעלת עומס העבודה של האצווה. הקובץ WAREHOUSE_FOLDER נמצא בדלי הזה.
- ‫--version: Managed Service for Apache Spark runtime version 2.2 או גרסה מתקדמת יותר.
- ‫SUBNET_NAME: השם של תת-רשת VPC ב-REGION. אם לא מציינים את הדגל הזה, Managed Service for Apache Spark בוחר את רשת המשנה default באזור הסשן. ‫Managed Service for Apache Spark מאפשר גישה פרטית ל-Google‏ (PGA) ברשת המשנה. למידע על דרישות הקישוריות לרשת, אפשר לעיין במאמר בנושא הגדרת הרשת של Managed Service for Apache Spark.
- ‫LOCATION: מיקום נתמך ב-BigQuery. מיקום ברירת המחדל הוא "US".
- ‫--properties נכסים מסוג קטלוג.
צפייה במטא-נתונים של טבלה ב-BigQuery
1. במסוף Google Cloud , עוברים לדף BigQuery.
  
  כניסה ל-BigQuery Studio
2. צפייה במטא-נתונים של טבלה בפורמט Iceberg.

PySpark

בשלבים הבאים מוסבר איך להריץ עומס עבודה של קובץ PySpark ב-Managed Service for Apache Spark כדי ליצור טבלת Iceberg עם מטא-נתונים של הטבלה שמאוחסנים בקטלוג של Lakehouse runtime.

מעתיקים את קוד PySpark הבא באופן מקומי או ב-Cloud Shell לקובץ iceberg-table.py.

from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("iceberg-table-example").getOrCreate()

catalog = "CATALOG_NAME"
namespace = "NAMESPACE"

spark.sql(f"USE `{catalog}`;")
spark.sql(f"CREATE NAMESPACE IF NOT EXISTS `{namespace}`;")
spark.sql(f"USE `{namespace}`;")

# Create table and display schema
spark.sql("DROP TABLE IF EXISTS example_iceberg_table")
spark.sql("CREATE TABLE example_iceberg_table (id int, data string) USING ICEBERG") 
spark.sql("DESCRIBE example_iceberg_table;")

# Insert table data.
spark.sql("INSERT INTO example_iceberg_table VALUES (1, 'first row');")

# Alter table, then display schema. 
spark.sql("ALTER TABLE example_iceberg_table ADD COLUMNS (newDoubleCol double);")
spark.sql("DESCRIBE example_iceberg_table;")