קטלוג Apache Iceberg בהתאמה אישית ל-BigQuery מאפשר לכם לחבר מנועי Apache Spark ו-Apache Flink באמצעות Apache Iceberg מגרסה 1.10 ואילך לקטלוג הייעודי לזמן ריצה של Lakehouse. ב-Lakehouse ללא גבולות, השילוב הזה מספק שכבת מטא-נתונים משותפת אחת לניהול פורמטים של טבלאות פתוחות באשכולות של Managed Service for Apache Spark ובManaged Service for Apache Spark ללא שרת (serverless).
לפני שמתחילים
- מפעילים את החיוב בפרויקט Google Cloud . כך בודקים אם החיוב מופעל בפרויקט
מפעילים את BigQuery API ואת Dataproc API.
התפקידים הנדרשים
כדי לקבל את ההרשאות שדרושות להגדרת הקטלוג הייעודי לזמן ריצה של Lakehouse, בקשו מהאדמין להקצות לכם את התפקידים הבאים ב-IAM:
-
יצירת אשכול של Managed Service for Apache Spark:
Dataproc Worker (
roles/dataproc.worker) בחשבון השירות שמוגדר כברירת מחדל של Compute Engine בפרויקט -
יוצרים טבלאות בקטלוג הייעודי לזמן ריצה של Lakehouse:
- Dataproc Worker (
roles/dataproc.worker) בחשבון השירות של Managed Service for Apache Spark VM בפרויקט - BigQuery Data Editor (
roles/bigquery.dataEditor) בחשבון השירות של Managed Service for Apache Spark VM בפרויקט - Storage Object User (
roles/storage.objectUser) בחשבון השירות של המכונה הווירטואלית של Managed Service for Apache Spark בפרויקט
- Dataproc Worker (
-
שליחת שאילתות לטבלאות בקטלוג הייעודי לזמן ריצה של Lakehouse:
- BigQuery Data Viewer (
roles/bigquery.dataViewer) בפרויקט - BigQuery User (
roles/bigquery.user) בפרויקט - צפייה באובייקטים באחסון (
roles/storage.objectViewer) בפרויקט
- BigQuery Data Viewer (
להסבר על מתן תפקידים, ראו איך מנהלים את הגישה ברמת הפרויקט, התיקייה והארגון.
יכול להיות שאפשר לקבל את ההרשאות הנדרשות גם באמצעות תפקידים בהתאמה אישית או תפקידים מוגדרים מראש.
הגדרת המטא-נתונים באמצעות אשכולות של Managed Service for Apache Spark
אתם יכולים להגדיר את הקטלוג הייעודי לזמן ריצה של Lakehouse עם אשכולות של Managed Service for Apache Spark באמצעות Spark או Flink:
Spark
מגדירים אשכול חדש. כדי ליצור אשכול חדש, מריצים את הפקודה הבאה של
gcloud dataproc clusters create, שמכילה את ההגדרות שצריך כדי להשתמש בקטלוג הייעודי לזמן ריצה של Lakehouse:gcloud dataproc clusters create CLUSTER_NAME \ --project=PROJECT_ID \ --region=LOCATION \ --single-node
מחליפים את מה שכתוב בשדות הבאים:
-
CLUSTER_NAME: שם לאשכול. -
PROJECT_ID: מזהה הפרויקט ב- Google Cloud שבו יוצרים את האשכול. -
LOCATION: האזור ב-Compute Engine שבו יוצרים את האשכול.
-
שולחים משימת Spark באחת מהשיטות הבאות:
Google Cloud CLI
gcloud dataproc jobs submit spark-sql \ --project=PROJECT_ID \ --cluster=CLUSTER_NAME \ --region=REGION \ --jars=https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-spark-runtime-3.5_2.12/1.10.0/iceberg-spark-runtime-3.5_2.12-1.10.0.jar,https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-bigquery/1.10.0/iceberg-bigquery-1.10.0.jar,https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-gcp-bundle/1.10.0/iceberg-gcp-bundle-1.10.0.jar,https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-gcp/1.10.0/iceberg-gcp-1.10.0.jar \ --properties=spark.sql.catalog.CATALOG_NAME=org.apache.iceberg.spark.SparkCatalog,\ spark.sql.catalog.CATALOG_NAME.type=bigquery,\ spark.sql.catalog.CATALOG_NAME.gcp.bigquery.project-id=PROJECT_ID,\ spark.sql.catalog.CATALOG_NAME.gcp.bigquery.location=LOCATION,\ spark.sql.catalog.CATALOG_NAME.warehouse=WAREHOUSE_DIRECTORY \ --execute="SPARK_SQL_COMMAND"
מחליפים את מה שכתוב בשדות הבאים:
-
PROJECT_ID: מזהה הפרויקט ב- Google Cloud שמכיל את האשכול. -
CLUSTER_NAME: שם האשכול שבו אתם משתמשים כדי להריץ את משימת Spark SQL. -
REGION: האזור ב-Compute Engine שבו נמצא האשכול. -
LOCATION: המיקום של המשאבים ב-BigQuery. -
CATALOG_NAME: השם של קטלוג Spark שבו משתמשים במשימת ה-SQL. -
WAREHOUSE_DIRECTORY: התיקייה ב-Cloud Storage שמכילה את מחסן הנתונים. הערך הזה מתחיל ב-gs://. -
SPARK_SQL_COMMAND: שאילתת Spark SQL שרוצים להריץ. השאילתה הזו כוללת את הפקודות ליצירת המשאבים. לדוגמה, כדי ליצור מרחב שמות וטבלה.
spark-sql CLI
נכנסים לדף VM instances במסוף Google Cloud .
כדי להתחבר למכונה וירטואלית של אשכול, לוחצים על SSH בשורה שבה מופיע השם של המכונה הווירטואלית הראשית של האשכול, שהוא שם האשכול ואחריו הסיומת
-m. הפלט אמור להיראות כך:Connected, host fingerprint: ssh-rsa ... Linux cluster-1-m 3.16.0-0.bpo.4-amd64 ... ... example-cluster@cluster-1-m:~$בטרמינל, מריצים את פקודת האתחול הבאה של הקטלוג הייעודי לזמן ריצה של Lakehouse:
spark-sql \ --jars https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-spark-runtime-3.5_2.12/1.10.0/iceberg-spark-runtime-3.5_2.12-1.10.0.jar,https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-bigquery/1.10.0/iceberg-bigquery-1.10.0.jar,https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-gcp-bundle/1.10.0/iceberg-gcp-bundle-1.10.0.jar,https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-gcp/1.10.0/iceberg-gcp-1.10.0.jar \ --conf spark.sql.catalog.CATALOG_NAME=org.apache.iceberg.spark.SparkCatalog \ --conf spark.sql.catalog.CATALOG_NAME.type=bigquery \ --conf spark.sql.catalog.CATALOG_NAME.gcp.bigquery.project-id=PROJECT_ID \ --conf spark.sql.catalog.CATALOG_NAME.gcp.bigquery.location=LOCATION \ --conf spark.sql.catalog.CATALOG_NAME.warehouse=WAREHOUSE_DIRECTORY
מחליפים את מה שכתוב בשדות הבאים:
-
CATALOG_NAME: השם של קטלוג Spark שבו אתם משתמשים במשימת ה-SQL. -
PROJECT_ID: מזהה הפרויקט ב- Google Cloud של הקטלוג הייעודי לזמן ריצה ב-Lakehouse, שאליו מקושר הקטלוג של Spark. -
LOCATION: המיקום של הקטלוג הייעודי לזמן ריצה של Lakehouse, ב- Google Cloud . -
WAREHOUSE_DIRECTORY: התיקייה ב-Cloud Storage שמכילה את מחסן הנתונים. הערך הזה מתחיל ב-gs://.
אחרי שמתחברים לאשכול, במסוף Spark מופיעה ההנחיה
spark-sql, שבה אפשר להשתמש כדי לשלוח משימות Spark.spark-sql (default)>-
-
Flink
- יוצרים אשכול Managed Service for Apache Spark כשרכיב Flink האופציונלי מופעל, ומוודאים שמשתמשים בגרסת אימג'
2.2ואילך. נכנסים לדף VM instances במסוף Google Cloud .
ברשימת המכונות הווירטואליות, לוחצים על SSH כדי להתחבר למכונה הווירטואלית הראשית של האשכול, שמופיעה בתור שם האשכול ואחריו הסיומת
-m.מגדירים את הפלאגין של הקטלוג בהתאמה אישית של Apache Iceberg לקטלוג הייעודי לזמן ריצה של Lakehouse:
FLINK_VERSION=1.20 ICEBERG_VERSION=1.10.0 cd /usr/lib/flink sudo wget -c https://repo.maven.apache.org/maven2/org/apache/iceberg/iceberg-flink-runtime-${FLINK_VERSION}/${ICEBERG_VERSION}/iceberg-flink-runtime-${FLINK_VERSION}-${ICEBERG_VERSION}.jar -P lib sudo wget -c https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-bigquery/${ICEBERG_VERSION}/iceberg-bigquery-${ICEBERG_VERSION}.jar -P lib sudo wget -c https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-gcp-bundle/${ICEBERG_VERSION}/iceberg-gcp-bundle-${ICEBERG_VERSION}.jar -P lib sudo wget -c https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-gcp/${ICEBERG_VERSION}/iceberg-gcp-${ICEBERG_VERSION}.jar -P lib
מתחילים את הסשן של Flink ב-YARN:
HADOOP_CLASSPATH=`hadoop classpath` sudo bin/yarn-session.sh -nm flink-dataproc -d sudo bin/sql-client.sh embedded \ -s yarn-session
יוצרים קטלוג ב-Flink:
CREATE CATALOG CATALOG_NAME WITH ( 'type'='iceberg', 'warehouse'='WAREHOUSE_DIRECTORY', 'catalog-impl'='org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog', 'gcp.bigquery.project-id'='PROJECT_ID', 'gcp.bigquery.location'='LOCATION' );
מחליפים את מה שכתוב בשדות הבאים:
-
CATALOG_NAME: המזהה של קטלוג Flink, שמקושר לקטלוג ייעודי לזמן ריצה של Lakehouse. -
WAREHOUSE_DIRECTORY: נתיב הבסיס של ספריית המחסן (התיקייה ב-Cloud Storage שבה Flink יוצר קבצים). הערך הזה מתחיל ב-gs://. -
PROJECT_ID: מזהה הפרויקט של הקטלוג הייעודי לזמן ריצה של Lakehouse, שאליו מקושר הקטלוג של Flink. -
LOCATION: המיקום של המשאבים ב-BigQuery.
-
הסשן של Flink מחובר עכשיו לקטלוג הייעודי לזמן ריצה של Lakehouse, ואתם יכולים להריץ פקודות SQL של Flink.
ניהול משאבים של קטלוג ייעודי לזמן ריצה ב-Lakehouse
עכשיו, אחרי שהתחברתם לקטלוג הייעודי לזמן ריצה של Lakehouse, אתם יכולים ליצור ולראות משאבים על סמך המטא-נתונים שמאוחסנים בקטלוג הייעודי לזמן ריצה של Lakehouse.
לדוגמה, נסו להריץ את הפקודות הבאות בסשן האינטראקטיבי של Flink SQL כדי ליצור מסד נתונים וטבלה של Apache Iceberg.
שימוש בקטלוג Apache Iceberg בהתאמה אישית:
USE CATALOG CATALOG_NAME;
מחליפים את
CATALOG_NAMEבמזהה הקטלוג של Flink.יוצרים מסד נתונים, שיוצר מערך נתונים ב-BigQuery:
CREATE DATABASE IF NOT EXISTS DATABASE_NAME;
מחליפים את
DATABASE_NAMEבשם של מסד הנתונים החדש.משתמשים במסד הנתונים שיצרתם:
USE DATABASE_NAME;
יוצרים טבלת Apache Iceberg. הקוד הבא יוצר טבלת מכירות לדוגמה:
CREATE TABLE IF NOT EXISTS ICEBERG_TABLE_NAME ( order_number BIGINT, price DECIMAL(32,2), buyer ROW<first_name STRING, last_name STRING>, order_time TIMESTAMP(3) );
מחליפים את
ICEBERG_TABLE_NAMEבשם של הטבלה החדשה.הצגת המטא-נתונים של הטבלה:
DESCRIBE EXTENDED ICEBERG_TABLE_NAME;
מציגים את הטבלאות במסד הנתונים:
SHOW TABLES;
הטמעת נתונים בטבלה
אחרי שיוצרים טבלת Apache Iceberg בקטע הקודם, אפשר להשתמש ב-Flink DataGen כמקור נתונים כדי להטמיע נתונים בזמן אמת בטבלה. השלבים הבאים הם דוגמה לתהליך העבודה הזה:
יוצרים טבלה זמנית באמצעות DataGen:
CREATE TEMPORARY TABLE DATABASE_NAME.TEMP_TABLE_NAME WITH ( 'connector' = 'datagen', 'rows-per-second' = '10', 'fields.order_number.kind' = 'sequence', 'fields.order_number.start' = '1', 'fields.order_number.end' = '1000000', 'fields.price.min' = '0', 'fields.price.max' = '10000', 'fields.buyer.first_name.length' = '10', 'fields.buyer.last_name.length' = '10' ) LIKE DATABASE_NAME.ICEBERG_TABLE_NAME (EXCLUDING ALL);
מחליפים את מה שכתוב בשדות הבאים:
-
DATABASE_NAME: שם מסד הנתונים שבו תאוחסן הטבלה הזמנית. -
TEMP_TABLE_NAME: שם לטבלה הזמנית. -
ICEBERG_TABLE_NAME: השם של טבלת Apache Iceberg שיצרתם בקטע הקודם.
-
מגדירים את המקביליות ל-1:
SET 'parallelism.default' = '1';
מגדירים את המרווח בין נקודות הבדיקה:
SET 'execution.checkpointing.interval' = '10second';
מגדירים את נקודת הציון:
SET 'state.checkpoints.dir' = 'hdfs:///flink/checkpoints';
מפעילים את משימת הסטרימינג בזמן אמת:
INSERT INTO ICEBERG_TABLE_NAME SELECT * FROM TEMP_TABLE_NAME;
הפלט אמור להיראות כך:
[INFO] Submitting SQL update statement to the cluster... [INFO] SQL update statement has been successfully submitted to the cluster: Job ID: 0de23327237ad8a811d37748acd9c10b
כדי לבדוק את הסטטוס של משימת הסטרימינג:
נכנסים לדף Clusters במסוף Google Cloud .
בוחרים את האשכול.
לוחצים על הכרטיסייה ממשקי אינטרנט.
לוחצים על הקישור YARN ResourceManager.
בממשק YARN ResourceManager, מאתרים את הסשן של Flink ולוחצים על הקישור ApplicationMaster בקטע Tracking UI.
בעמודה Status, מוודאים שסטטוס המשימה הוא Running.
מריצים שאילתה על נתונים בסטרימינג בלקוח Flink SQL:
SELECT * FROM ICEBERG_TABLE_NAME /*+ OPTIONS('streaming'='true', 'monitor-interval'='3s')*/ ORDER BY order_time desc LIMIT 20;
הרצת שאילתות על נתונים בסטרימינג ב-BigQuery:
SELECT * FROM `DATABASE_NAME.ICEBERG_TABLE_NAME` ORDER BY order_time desc LIMIT 20;
מסיימים את משימת הסטרימינג בלקוח Flink SQL:
STOP JOB 'JOB_ID';
מחליפים את
JOB_IDבמזהה המשימה שהוצג בפלט כשיצרתם את משימת הסטרימינג.
הגדרת המטא-נתונים באמצעות Managed Service for Apache Spark ללא שרת (serverless)
אתם יכולים להגדיר את הקטלוג הייעודי לזמן ריצה של Lakehouse עם Managed Service for Apache Spark בלי שרת (serverless), באמצעות Spark SQL או PySpark.
Spark SQL
יוצרים קובץ SQL עם פקודות Spark SQL שרוצים להריץ בקטלוג הייעודי לזמן ריצה של Lakehouse. לדוגמה, הפקודה הבאה יוצרת מרחב שמות וטבלה:
SET `spark.sql.catalog.CATALOG_NAME`=`org.apache.iceberg.spark.SparkCatalog`; SET `spark.sql.catalog.CATALOG_NAME.type`=`bigquery`; SET `spark.sql.catalog.CATALOG_NAME.gcp.bigquery.project-id`=`PROJECT_ID`; SET `spark.sql.catalog.CATALOG_NAME.gcp.bigquery.location`=`LOCATION`; SET `spark.sql.catalog.CATALOG_NAME.warehouse`=`WAREHOUSE_DIRECTORY`; CREATE NAMESPACE `CATALOG_NAME`.NAMESPACE_NAME; CREATE TABLE `CATALOG_NAME`.NAMESPACE_NAME.TABLE_NAME (id int, data string) USING ICEBERG LOCATION 'WAREHOUSE_DIRECTORY';
מחליפים את מה שכתוב בשדות הבאים:
-
CATALOG_NAME: שם הקטלוג שמפנה לטבלת Spark. -
NAMESPACE_NAME: השם של מרחב השמות שמפנה לטבלת Spark. -
TABLE_NAME: שם לטבלת Spark. -
WAREHOUSE_DIRECTORY: ה-URI של התיקייה ב-Cloud Storage שבה מאוחסן מחסן הנתונים.
-
מריצים את הפקודה הבאה של
gcloud dataproc batches submit spark-sqlכדי לשלוח משימת Spark SQL באצווה:gcloud dataproc batches submit spark-sql SQL_SCRIPT_PATH \ --project=PROJECT_ID \ --region=REGION \ --subnet=projects/PROJECT_ID/regions/REGION/subnetworks/SUBNET_NAME \ --deps-bucket=BUCKET_PATH \ --jars=https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-spark-runtime-3.5_2.12/1.10.0/iceberg-spark-runtime-3.5_2.12-1.10.0.jar,https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-bigquery/1.10.0/iceberg-bigquery-1.10.0.jar,https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-gcp-bundle/1.10.0/iceberg-gcp-bundle-1.10.0.jar,https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-gcp/1.10.0/iceberg-gcp-1.10.0.jar
מחליפים את מה שכתוב בשדות הבאים:
-
SQL_SCRIPT_PATH: הנתיב לקובץ ה-SQL שבו משתמשת המשימה באצווה. -
PROJECT_ID: מזהה הפרויקט ב- Google Cloud שבו רוצים להריץ את המשימה באצווה. -
REGION: האזור שבו פועל עומס העבודה. -
SUBNET_NAME(אופציונלי): השם של רשת משנה של VPC ב-REGIONשעומדת בדרישות של רשת המשנה של הסשן. -
BUCKET_PATH: המיקום של הקטגוריה ב-Cloud Storage להעלאת יחסי התלות של עומס העבודה. ה-WAREHOUSE_DIRECTORYנמצא בקטגוריה הזו. לא צריך את הקידומתgs://של ה-URI של הקטגוריה. אפשר לציין את הנתיב או השם של הקטגוריה, לדוגמהmybucketname1. -
LOCATION: המיקום שבו תתבצע המשימה באצווה.
מידע נוסף על שליחת משימות באצווה של Spark זמין במאמר הרצת עומס עבודה באצווה של Spark.
-
PySpark
יוצרים קובץ Python עם פקודות PySpark שרוצים להריץ בקטלוג הייעודי לזמן ריצה של Lakehouse.
לדוגמה, הפקודה הבאה מגדירה סביבת Spark לאינטראקציה עם טבלאות Apache Iceberg שמאוחסנות בקטלוג הייעודי לזמן ריצה של Lakehouse. לאחר מכן, הפקודה יוצרת מרחב שמות חדש וטבלת Apache Iceberg בתוך מרחב השמות הזה.
from pyspark.sql import SparkSession spark = SparkSession.builder \ .appName("Lakehouse runtime catalog Iceberg") \ .config("spark.sql.catalog.CATALOG_NAME", "org.apache.iceberg.spark.SparkCatalog") \ .config("spark.sql.catalog.CATALOG_NAME.type", "bigquery") \ .config("spark.sql.catalog.CATALOG_NAME.gcp.bigquery.project-id", "PROJECT_ID") \ .config("spark.sql.catalog.CATALOG_NAME.gcp.bigquery.location", "LOCATION") \ .config("spark.sql.catalog.CATALOG_NAME.warehouse", "WAREHOUSE_DIRECTORY") \ .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';")
מחליפים את מה שכתוב בשדות הבאים:
-
PROJECT_ID: מזהה הפרויקט ב- Google Cloud שבו רוצים להריץ את המשימה באצווה. -
LOCATION: המיקום שבו נמצאים המשאבים של BigQuery. -
CATALOG_NAME: שם הקטלוג שמפנה לטבלת Spark. -
TABLE_NAME: שם לטבלת Spark. -
WAREHOUSE_DIRECTORY: ה-URI של התיקייה ב-Cloud Storage שבה מאוחסן מחסן הנתונים. -
NAMESPACE_NAME: השם של מרחב השמות שמפנה לטבלת Spark.
-
שולחים את משימת האצווה באמצעות הפקודה
gcloud dataproc batches submit pyspark:gcloud dataproc batches submit pyspark PYTHON_SCRIPT_PATH \ --version=2.2 \ --project=PROJECT_ID \ --region=REGION \ --subnet=projects/PROJECT_ID/regions/REGION/subnetworks/SUBNET_NAME \ --deps-bucket=BUCKET_PATH \ --jars=https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-spark-runtime-3.5_2.12/1.10.0/iceberg-spark-runtime-3.5_2.12-1.10.0.jar,https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-bigquery/1.10.0/iceberg-bigquery-1.10.0.jar,https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-gcp-bundle/1.10.0/iceberg-gcp-bundle-1.10.0.jar,https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-gcp/1.10.0/iceberg-gcp-1.10.0.jar
מחליפים את מה שכתוב בשדות הבאים:
-
PYTHON_SCRIPT_PATH: הנתיב לסקריפט Python שמשמש את המשימה באצווה. -
PROJECT_ID: מזהה הפרויקט ב- Google Cloud שבו רוצים להריץ את המשימה באצווה. -
REGION: האזור שבו פועל עומס העבודה. -
BUCKET_PATH: המיקום של הקטגוריה ב-Cloud Storage להעלאת יחסי התלות של עומס העבודה. לא צריך להוסיף את הקידומתgs://ל-URI של הקטגוריה. אפשר לציין את הנתיב או את השם של הקטגוריה, לדוגמה,mybucketname1.
למידע נוסף על שליחת משימות אצווה של PySpark, קראו את ההפניה ל-gcloud של PySpark.
-