רכיב Iceberg אופציונלי של Managed Service for Apache Spark

הרכיב האופציונלי Iceberg מאפשר לנהל טבלאות של data lakehouse עם טרנזקציות ACID ופיתוח סכימה באשכולות של Managed Service for Apache Spark. אפשר להתקין את Iceberg במהלך יצירת האשכול באמצעות התכונה Optional components כדי לשלוח שאילתות ולעדכן טבלאות בעומסי עבודה של Spark ו-Hive.

התכונות העיקריות של Iceberg

התכונות של Iceberg כוללות:

• התפתחות הסכימה: הוספה, הסרה או שינוי שם של עמודות בלי לשכתב את כל הטבלה.
• Time travel: שליחת שאילתות על תמונות מצב היסטוריות של טבלאות למטרות ביקורת או החזרה למצב קודם.
• חלוקה למחיצות מוסתרות: אופטימיזציה של פריסת הנתונים לשאילתות מהירות יותר בלי לחשוף את פרטי המחיצות למשתמשים.
• טרנזקציות ACID: מבטיחות עקביות בנתונים ומונעות התנגשויות.

מאפיינים שקשורים ל-Iceberg

כשיוצרים אשכול Managed Service for Apache Spark with Iceberg, המאפיינים הבאים של Spark ו-Hive מוגדרים לעבודה עם Iceberg.

גרסת תמונה 2.x

המאפיינים הבאים של Spark ו-Hive מוגדרים באשכולות Managed Service for Apache Spark בגרסה `2.x` כשמתקינים את רכיב Iceberg:

קובץ תצורה	מאפיין (property)	ערך ברירת המחדל
/etc/spark/conf/spark-defaults.conf	spark.sql.extensions	org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
	spark.driver.extraClassPath	/usr/lib/iceberg/lib/iceberg-spark-runtime-spark-version_scala-version.jar
	spark.executor.extraClassPath	/usr/lib/iceberg/lib/iceberg-spark-runtime-spark-version_scala-version.jar
/etc/hive/conf/hive-site.xml	hive.aux.jars.path	file:///usr/lib/iceberg/lib/iceberg-hive-runtime.jar
	iceberg.engine.hive.enabled	true

גרסת תמונה 3.0

המאפיינים הבאים של Spark ו-Hive מוגדרים באשכולות Managed Service for Apache Spark בגרסה 3.0 כשמתקינים את רכיב Iceberg:

קובץ תצורה	מאפיין (property)	ערך ברירת המחדל
/etc/spark/conf/spark-defaults.conf	spark.sql.extensions	org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
	spark.sql.catalog.default_iceberg	org.apache.iceberg.spark.SparkCatalog
	spark.sql.catalog.default_iceberg.type	rest
	spark.sql.catalog.default_iceberg.uri	env:ICEBERG_REST_CATALOG_URI
/etc/hive/conf/hive-site.xml	iceberg.engine.hive.enabled	true

התקנת הרכיב האופציונלי Iceberg

אפשר להתקין את רכיב Iceberg באשכולות של Managed Service for Apache Spark שנוצרו עם גרסאות אימג' 2.2.47 ואילך. בדפים של רשימת גרסאות האימג' של אשכולות Managed Service for Apache Spark מוצגת גרסת הרכיב של Iceberg שכלולה בגרסאות האימג' האחרונות של אשכולות Managed Service for Apache Spark.

הערה: גרסת התמונה 3.0 של Managed Service for Apache Spark כוללת את Hive 4.2, שתומך ב-Iceberg דרך קטלוג ה-REST של Iceberg.

מסוףGoogle Cloud

כדי ליצור אשכול של Managed Service for Apache Spark שמתקין את רכיב Iceberg, מבצעים את השלבים הבאים במסוף Google Cloud :

1. פותחים את הדף Create cluster.
2. לוחצים על הגדרות נוספות כדי להרחיב את הקטע.
3. עורכים את Optional components.
4. בחלונית שנפתחת, מסמנים את התיבה Iceberg.
5. לוחצים על Save.

‫CLI של gcloud

כדי ליצור אשכול של Managed Service for Apache Spark שמתקין את רכיב Iceberg, משתמשים בפקודה gcloud dataproc clusters create עם הדגל --optional-components.

gcloud dataproc clusters create CLUSTER_NAME \
    --region=REGION \
    --optional-components=ICEBERG \
     other flags ...

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

• ‫CLUSTER_NAME: השם החדש של האשכול.
• ‫REGION: האזור של האשכול.

‫API בארכיטקטורת REST

כדי ליצור אשכול Managed Service for Apache Spark שמתקין את הרכיב האופציונלי של Iceberg, צריך לציין את Iceberg SoftwareConfig.Component כחלק מבקשת clusters.create.

שימוש בטבלאות Iceberg עם Spark ו-Hive

אחרי שיוצרים אשכול Managed Service for Apache Spark שמותקן בו הרכיב האופציונלי Iceberg, אפשר להשתמש ב-Spark וב-Hive כדי לקרוא ולכתוב נתונים בטבלאות Iceberg.

Spark

הגדרת סשן Spark ל-Iceberg

כדי להפעיל את התוספים של Iceberg ל-Spark ולהגדיר את הקטלוג של Spark לשימוש בטבלאות Iceberg, אתם יכולים להשתמש בפקודת ה-CLI של gcloud באופן מקומי, או בלולאות REPL (קריאה-הערכה-הדפסה) של spark-shell או pyspark שפועלות בצומת הראשי של אשכול Dataproc.

gcloud

מריצים את הדוגמה הבאה של ה-CLI של gcloud בחלון טרמינל מקומי או ב-Cloud Shell כדי לשלוח משימת Spark ולהגדיר את מאפייני Spark, כדי להגדיר את הסשן של Spark ל-Iceberg.

gcloud dataproc jobs submit spark  \
    --cluster=CLUSTER_NAME \
    --region=REGION \
    --properties="spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions" \
    --properties="spark.sql.catalog.CATALOG_NAME=org.apache.iceberg.spark.SparkCatalog" \
    --properties="spark.sql.catalog.CATALOG_NAME.type=hadoop" \
    --properties="spark.sql.catalog.CATALOG_NAME.warehouse=gs://BUCKET/FOLDER" \
     other flags ...

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

• ‫CLUSTER_NAME: שם האשכול.
• REGION: האזור של Compute Engine.
• ‫CATALOG_NAME: שם הקטלוג של Iceberg.
• ‫BUCKET ו-FOLDER: המיקום של קטלוג Iceberg ב-Cloud Storage.

spark-shell

כדי להגדיר סשן של Spark ל-Iceberg באמצעות spark-shell REPL באשכול של Managed Service for Apache Spark:

1. משתמשים ב-SSH כדי להתחבר לצומת הראשי של האשכול ב-Managed Service for Apache Spark.

2. מריצים את הפקודה הבאה במסוף של סשן ה-SSH כדי להגדיר את סשן Spark ל-Iceberg.

spark-shell \
    --conf "spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions" \
    --conf "spark.sql.catalog.CATALOG_NAME=org.apache.iceberg.spark.SparkCatalog" \
    --conf "spark.sql.catalog.CATALOG_NAME.type=hadoop" \
    --conf "spark.sql.catalog.CATALOG_NAME.warehouse=gs://BUCKET/FOLDER"

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

• ‫CLUSTER_NAME: שם האשכול.
• REGION: האזור של Compute Engine.
• ‫CATALOG_NAME: שם הקטלוג של Iceberg.
• ‫BUCKET ו-FOLDER: המיקום של קטלוג Iceberg ב-Cloud Storage.

מעטפת pyspark

כדי להגדיר סשן של Spark ל-Iceberg באמצעות pyspark REPL באשכול של Managed Service for Apache Spark:

1. משתמשים ב-SSH כדי להתחבר לצומת הראשי של האשכול ב-Managed Service for Apache Spark.

2. מריצים את הפקודה הבאה במסוף של סשן ה-SSH כדי להגדיר את סשן Spark ל-Iceberg:

pyspark \
    --conf "spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions" \
    --conf "spark.sql.catalog.CATALOG_NAME=org.apache.iceberg.spark.SparkCatalog" \
    --conf "spark.sql.catalog.CATALOG_NAME.type=hadoop" \
    --conf "spark.sql.catalog.CATALOG_NAME.warehouse=gs://BUCKET/FOLDER"

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

• ‫CLUSTER_NAME: שם האשכול.
• REGION: האזור של Compute Engine.
• ‫CATALOG_NAME: שם הקטלוג של Iceberg.
• ‫BUCKET ו-FOLDER: המיקום של קטלוג Iceberg ב-Cloud Storage.

הערה: כדי להגדיר את Spark כך שישתמש ב-Hive Metastore כקטלוג Iceberg ב-Managed Service for Apache Spark מגרסה 3.0 ואילך, משתמשים במאפיינים הבאים: * spark.sql.catalog.<CATALOG_NAME>.type=rest * spark.sql.catalog.<CATALOG_NAME>.uri=env:ICEBERG_REST_CATALOG_URI * (לאשכולות שבהם Kerberos מופעל) spark.sql.catalog.<CATALOG_NAME>.token=env:ICEBERG_REST_CATALOG_TOKEN

כתיבת נתונים בטבלת Iceberg

אתם יכולים לכתוב נתונים בטבלת Iceberg באמצעות Spark. קטעי הקוד הבאים יוצרים DataFrame עם נתונים לדוגמה, יוצרים טבלת Iceberg ב-Cloud Storage ואז כותבים את הנתונים בטבלת Iceberg.

PySpark

# Create a DataFrame with sample data.
data = spark.createDataFrame([(1, "Alice"), (2, "Bob")], ["id", "name"])

# Create an Iceberg table in Cloud Storage.
spark.sql("""CREATE TABLE IF NOT EXISTS CATALOG_NAME.NAMESPACE.TABLE_NAME (
    id integer,
    name string)
USING iceberg
LOCATION 'gs://BUCKET/FOLDER/NAMESPACE/TABLE_NAME'""")

# Write the DataFrame to the Iceberg table in Cloud Storage.
data.writeTo("CATALOG_NAME.NAMESPACE.TABLE_NAME").append()

Scala

// Create a DataFrame with sample data.
val data = Seq((1, "Alice"), (2, "Bob")).toDF("id", "name")

// Create an Iceberg table in Cloud Storage.
spark.sql("""CREATE TABLE IF NOT EXISTS CATALOG_NAME.NAMESPACE.TABLE_NAME (
    id integer,
    name string)
USING iceberg
LOCATION 'gs://BUCKET/FOLDER/NAMESPACE/TABLE_NAME'""")

// Write the DataFrame to the Iceberg table in Cloud Storage.
data.writeTo("CATALOG_NAME.NAMESPACE.TABLE_NAME").append()

קריאת נתונים מטבלת Iceberg

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

PySpark

# Read Iceberg table data into a DataFrame.
df = spark.read.format("iceberg").load("CATALOG_NAME.NAMESPACE.TABLE_NAME")
# Display the data.
df.show()

Scala

// Read Iceberg table data into a DataFrame.
val df = spark.read.format("iceberg").load("CATALOG_NAME.NAMESPACE.TABLE_NAME")

// Display the data.
df.show()

Spark SQL

SELECT * FROM CATALOG_NAME.NAMESPACE.TABLE_NAME

Hive

יצירת טבלת Iceberg ב-Hive

אזהרה: ב-Iceberg אין תמיכה ישירה ב-Hive 4.2. אפשר לקבל תמיכה באמצעות קטלוג REST של Iceberg שנתמך על ידי Hive. ההוראות הבאות לשימוש ב-HiveIcebergStorageHandler רלוונטיות רק לגרסאות 2.x של Managed Service for Apache Spark.

באשכולות של Managed Service for Apache Spark, ‏Hive מוגדר מראש לעבוד עם Iceberg.

כדי להריץ את קטעי הקוד שבקטע הזה, צריך לבצע את הפעולות הבאות:

1. משתמשים ב-SSH כדי להתחבר לצומת הראשי של אשכול Managed Service for Apache Spark.

2. מציגים את beeline בחלון הטרמינל של SSH.

   beeline -u jdbc:hive2://
   

אתם יכולים ליצור טבלת Iceberg עם או בלי מחיצות ב-Hive.

טבלה שלא חולקה למחיצות (Partitions)

יוצרים טבלת Iceberg לא מחולקת ב-Hive.

CREATE TABLE my_table (
  id INT,
  name STRING,
  created_at TIMESTAMP
) STORED BY 'org.apache.iceberg.mr.hive.HiveIcebergStorageHandler';

טבלה מחולקת למחיצות (Partitions)

כדי ליצור טבלת Iceberg מחולקת למחיצות ב-Hive, מציינים את עמודות המחיצה בסעיף PARTITIONED BY.

CREATE TABLE my_partitioned_table (
  id INT,
  name STRING
) PARTITIONED BY (date_sk INT)
STORED BY 'org.apache.iceberg.mr.hive.HiveIcebergStorageHandler';

הוספת נתונים לטבלת Iceberg ב-Hive

אפשר להוסיף נתונים לטבלת Iceberg באמצעות הצהרות INSERT רגילות של Hive.

SET hive.execution.engine=mr;

INSERT INTO my_table
SELECT 1, 'Alice', current_timestamp();

מגבלות

• מנוע הביצוע MR ‏(MapReduce) נתמך רק בפעולות DML (שפת טיפול בנתונים).
• הפעלת MR הוצאה משימוש ב-Hive 3.1.3.

קריאת נתונים מטבלת Iceberg ב-Hive

כדי לקרוא נתונים מטבלת Iceberg, משתמשים בהצהרת SELECT.

SELECT * FROM my_table;

מחיקת טבלת Iceberg ב-Hive.

כדי להסיר טבלת Iceberg ב-Hive, משתמשים בהצהרה DROP TABLE.

DROP TABLE my_table;

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

• המדריך למתחילים של Iceberg