התקנת רכיב Delta Lake באשכול Dataproc

במאמר הזה מוסבר איך להתקין ולהגדיר את הרכיב האופציונלי Delta Lake באשכול Dataproc. כשהרכיב Delta Lake מופעל, הוא מתקין את הספריות הנדרשות ומגדיר את Spark ו-Hive כך שיפעלו עם Delta Lake.

מידע נוסף על רכיבים אופציונליים אחרים שזמינים ב-Dataproc מופיע במאמר רכיבים אופציונליים זמינים.

גרסאות תמונות Dataproc תואמות

אפשר להתקין את רכיב Delta Lake באשכולות Dataproc שנוצרו עם גרסת האימג' של Dataproc‏ 2.2.46 ומגרסאות אימג' מאוחרות יותר.

במאמר גרסאות נתמכות של Dataproc מפורטת גרסת הרכיב Delta Lake שכלולה בגרסאות של תמונות Dataproc.

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

כשיוצרים אשכול Dataproc עם הרכיב Delta Lake מופעל, מאפייני Spark הבאים מוגדרים לעבודה עם Delta Lake.

קובץ תצורה	מאפיין (property)	ערך ברירת המחדל
/etc/spark/conf/spark-defaults.conf	spark.sql.extensions	io.delta.sql.DeltaSparkSessionExtension
/etc/spark/conf/spark-defaults.conf	spark.sql.catalog.spark_catalog	org.apache.spark.sql.delta.catalog.DeltaCatalog

התקנת הרכיב

אפשר להתקין את הרכיב כשיוצרים אשכול Dataproc באמצעות מסוף Google Cloud ,‏ Google Cloud CLI או Dataproc API.

המסוף

1. נכנסים לדף Create a cluster ב-Dataproc במסוף Google Cloud .

   כניסה לדף Create a cluster

   החלונית הגדרת אשכול נבחרת.

2. בקטע Components, מתחת לOptional components, בוחרים באפשרות Delta Lake ובאפשרויות אחרות של רכיבים להתקנה באשכול.

‫CLI של gcloud

כדי ליצור אשכול Dataproc שכולל את רכיב Delta Lake, משתמשים בפקודה gcloud dataproc clusters create עם הדגל --optional-components.

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

הערות:

• ‫CLUSTER_NAME: מציינים את שם האשכול.
• ‫REGION: מציינים אזור של Compute Engine שבו האשכול ימוקם.

API ל-REST

אפשר לציין את הרכיב Delta Lake דרך Dataproc API באמצעות SoftwareConfig.Component כחלק מבקשת clusters.create.

דוגמאות לשימוש

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

טבלת Delta Lake

כתיבה לטבלת Delta Lake

אפשר להשתמש ב-Spark DataFrame כדי לכתוב נתונים לטבלת Delta Lake. בדוגמאות הבאות נוצר DataFrame עם נתונים לדוגמה, נוצרת טבלת my_delta_table Delta Lake ב-Cloud Storage, ואז הנתונים נכתבים בטבלת Delta Lake.

PySpark

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

# Create a Delta Lake table in Cloud Storage.
spark.sql("""CREATE TABLE IF NOT EXISTS my_delta_table (
    id integer,
    name string)
USING delta
LOCATION 'gs://delta-gcs-demo/example-prefix/default/my_delta_table'""")

# Write the DataFrame to the Delta Lake table in Cloud Storage.
data.writeTo("my_delta_table").append()

Scala

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

// Create a Delta Lake table in Cloud Storage.
spark.sql("""CREATE TABLE IF NOT EXISTS my_delta_table (
    id integer,
    name string)
USING delta
LOCATION 'gs://delta-gcs-demo/example-prefix/default/my_delta_table'""")

// Write the DataFrame to the Delta Lake table in Cloud Storage.
data.write.format("delta").mode("append").saveAsTable("my_delta_table")

Spark SQL

CREATE TABLE IF NOT EXISTS my_delta_table (
    id integer,
    name string)
USING delta
LOCATION 'gs://delta-gcs-demo/example-prefix/default/my_delta_table';

INSERT INTO my_delta_table VALUES ("1", "Alice"), ("2", "Bob");

קריאה מטבלת Delta Lake

בדוגמאות הבאות, הקוד קורא את my_delta_table ומציג את התוכן שלו.

PySpark

# Read the Delta Lake table into a DataFrame.
df = spark.table("my_delta_table")

# Display the data.
df.show()

Scala

// Read the Delta Lake table into a DataFrame.
val df = spark.table("my_delta_table")

// Display the data.
df.show()

Spark SQL

SELECT * FROM my_delta_table;

‫Hive עם Delta Lake

כתיבה לטבלת Delta ב-Hive.

רכיב Dataproc Delta Lake האופציונלי מוגדר מראש לעבודה עם טבלאות חיצוניות של Hive.

מידע נוסף זמין במאמר מחבר Hive.

מריצים את הדוגמאות בלקוח beeline.

beeline -u jdbc:hive2://

ליצור טבלת Spark Delta Lake.

צריך ליצור את טבלת Delta Lake באמצעות Spark לפני שטבלת Hive חיצונית תוכל להפנות אליה.

CREATE TABLE IF NOT EXISTS my_delta_table (
    id integer,
    name string)
USING delta
LOCATION 'gs://delta-gcs-demo/example-prefix/default/my_delta_table';

INSERT INTO my_delta_table VALUES ("1", "Alice"), ("2", "Bob");

יוצרים טבלה חיצונית ב-Hive.

SET hive.input.format=io.delta.hive.HiveInputFormat;
SET hive.tez.input.format=io.delta.hive.HiveInputFormat;

CREATE EXTERNAL TABLE deltaTable(id INT, name STRING)
STORED BY 'io.delta.hive.DeltaStorageHandler'
LOCATION 'gs://delta-gcs-demo/example-prefix/default/my_delta_table';

הערות:

• המחלקות io.delta.hive.DeltaStorageHandler מטמיעות את ממשקי ה-API של מקור הנתונים של Hive. הוא יכול לטעון טבלת Delta ולחלץ ממנה את המטא-נתונים. אם סכימת הטבלה בהצהרת CREATE TABLE לא עקבית עם המטא-נתונים הבסיסיים של Delta Lake, מוצגת שגיאה.

קריאה מטבלת Delta Lake ב-Hive.

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

SELECT * FROM deltaTable;

משחררים טבלה של Delta Lake.

כדי להשליך טבלת Delta, משתמשים בהצהרה DROP TABLE:

DROP TABLE deltaTable;