שימוש בקטלוג של Lakehouse runtime עם פרוצדורות מאוחסנות של Apache Spark

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

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

  1. מפעילים את החיוב בפרויקט Google Cloud . כך בודקים אם החיוב מופעל בפרויקט

  2. מפעילים את ממשקי ה-API של BigQuery ו-Dataproc.

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

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

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

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

  • יוצרים טבלאות קטלוג של Lakehouse בזמן ריצה ב-Apache Spark:
    • BigQuery Data Editor (roles/bigquery.dataEditor) בחשבון השירות של Spark Connection בפרויקט
    • Storage Object User (roles/storage.objectUser) בחשבון השירות של Spark Connection בפרויקט
  • שליחת שאילתות על טבלאות קטלוג של זמן ריצה של Lakehouse ב-BigQuery :

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

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

יצירה והפעלה של תהליך מאוחסן

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

  1. עוברים לדף BigQuery.

    כניסה ל-BigQuery

  2. בעורך השאילתות, מוסיפים את קוד לדוגמה הבא להצהרת CREATE PROCEDURE.

    CREATE OR REPLACE PROCEDURE
`PROJECT_ID.BQ_DATASET_ID.PROCEDURE_NAME`()
WITH CONNECTION `PROJECT_ID.REGION.SPARK_CONNECTION_ID` OPTIONS (engine='SPARK',
runtime_version='1.1',
properties=[("spark.sql.catalog.CATALOG_NAME.warehouse",
"WAREHOUSE_DIRECTORY"),
("spark.sql.catalog.CATALOG_NAME.gcp_location",
"LOCATION"),
("spark.sql.catalog.CATALOG_NAME.gcp_project",
"PROJECT_ID"),
("spark.sql.catalog.CATALOG_NAME",
"org.apache.iceberg.spark.SparkCatalog"),
("spark.sql.catalog.CATALOG_NAME.catalog-impl",
"org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog"),
("spark.jars.packages",
"org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.6.1")],
jar_uris=["gs://spark-lib/bigquery/iceberg-bigquery-catalog-1.6.1-1.0.2.jar"])
LANGUAGE python AS R"""
from pyspark.sql import SparkSession
spark = SparkSession \
.builder \
.appName("Lakehouse runtime catalog Iceberg") \
.getOrCreate()
spark.sql("USE CATALOG_NAME;")
spark.sql("CREATE NAMESPACE IF NOT EXISTS NAMESPACE_NAME;")
spark.sql("USE NAMESPACE_NAME;")
spark.sql("CREATE TABLE TABLE_NAME (id int, data string) USING ICEBERG LOCATION 'WAREHOUSE_DIRECTORY'")
spark.sql("DESCRIBE TABLE_NAME;")
spark.sql("INSERT INTO TABLE_NAME VALUES (1, \"first row\");")
spark.sql("SELECT * from TABLE_NAME;")
spark.sql("ALTER TABLE TABLE_NAME ADD COLUMNS (newDoubleCol double);")
spark.sql("DESCRIBE TABLE_NAME;")
""";
CALL `PROJECT_ID.BQ_DATASET_ID.PROCEDURE_NAME`();

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

    • ‫PROJECT_ID: מזהה הפרויקט ב- Google Cloud .
    • ‫BQ_DATASET_ID: המזהה של מערך הנתונים ב-BigQuery שמכיל את הפרוצדורה.
    • ‫PROCEDURE_NAME: השם של הפרוצדורה שיוצרים או מחליפים.
    • ‫REGION: המיקום של חיבור Apache Spark.
    • ‫LOCATION: המיקום של משאבי BigQuery.
    • ‫SPARK_CONNECTION_ID: המזהה של חיבור Apache Spark.
    • ‫CATALOG_NAME: השם של הקטלוג שבו אתם משתמשים.
    • ‫WAREHOUSE_DIRECTORY: ה-URI של תיקיית Cloud Storage שמכילה את מחסן הנתונים.
    • ‫NAMESPACE_NAME: מרחב השמות שבו אתם משתמשים.

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