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

הרכיב Apache Hudi מאפשר להריץ את Hudi ב-Managed Service for Apache Spark. כשמתקינים את הרכיב במהלך יצירת האשכול באמצעות התכונה Optional components, הוא פורס את ספריות Hudi ומגדיר את Spark ו-Hive באשכול כך שיעבדו עם Hudi.

גרסאות תואמות של Managed Service for Apache Spark

אפשר להתקין את רכיב Hudi באשכולות של Managed Service for Apache Spark שנוצרו באמצעות גרסאות האימג' הבאות של Managed Service for Apache Spark:

• 2.0.54+
• 2.1.2+
• 2.2.31+

הערה: Hudi לא זמין בגרסת האימג' 3.0 של Managed Service for Apache Spark.

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

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

קובץ תצורה	מאפיין (property)	ערך ברירת המחדל
/etc/spark/conf/spark-defaults.conf	spark.serializer	org.apache.spark.serializer.KryoSerializer
	spark.sql.catalog.spark_catalog	org.apache.spark.sql.hudi.catalog.HoodieCatalog
	spark.sql.extensions	org.apache.spark.sql.hudi.HoodieSparkSessionExtension
	spark.driver.extraClassPath	/usr/lib/hudi/lib/hudi-sparkspark-version-bundle_scala-version-hudi-version.jar
	spark.executor.extraClassPath	/usr/lib/hudi/lib/hudi-sparkspark-version-bundle_scala-version-hudi-version.jar
/etc/hive/conf/hive-site.xml	hive.aux.jars.path	file:///usr/lib/hudi/lib/hudi-hadoop-mr-bundle-version.jar

התקנת הרכיב

מתקינים את רכיב Hudi כשיוצרים אשכול Managed Service for Apache Spark.

בדפי גרסאות ההפצה של קובצי האימג' של Managed Service for Apache Spark מפורטות גרסאות הרכיבים של Hudi שכלולות בכל גרסת הפצה של קובץ אימג' של Managed Service for Apache Spark.

מסוףGoogle Cloud

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

‫CLI של gcloud

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

gcloud dataproc clusters create CLUSTER_NAME \
    --region=REGION \
    --optional-components=HUDI \
    --image-version=DATAPROC_VERSION \
    --properties=PROPERTIES

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

• ‫CLUSTER_NAME: חובה. השם החדש של האשכול.
• ‫REGION: חובה. האזור של האשכול.
• ‫DATAPROC_IMAGE: אופציונלי. אתם יכולים להשתמש בדגל האופציונלי הזה כדי לציין גרסת אימג' של Managed Service for Apache Spark שהיא לא ברירת המחדל (ראו גרסת אימג' של Managed Service for Apache Spark שמוגדרת כברירת מחדל).
• ‫PROPERTIES: אופציונלי. אפשר להשתמש בדגל האופציונלי הזה כדי להגדיר את המאפיינים של רכיב Hudi, שמצוינים באמצעות הקידומת של הקובץ hudi: (לדוגמה: properties=hudi:hoodie.datasource.write.table.type=COPY_ON_WRITE).
  • מאפיין גרסת הרכיב של Hudi: אפשר לציין את המאפיין dataproc:hudi.version. הערה: גרסת הרכיב של Hudi מוגדרת על ידי Managed Service for Apache Spark כך שתתאים לגרסת קובץ האימג' של האשכול ב-Managed Service for Apache Spark. אם מגדירים את המאפיין הזה, יכול להיות שהאשכול לא ייווצר אם הגרסה שצוינה לא תואמת לתמונת האשכול.
  • מאפייני Spark ו-Hive: כשיוצרים את האשכול, Managed Service for Apache Spark מגדיר את מאפייני Spark ו-Hive שקשורים ל-Hudi. לא צריך להגדיר אותם כשיוצרים את האשכול או שולחים משימות.

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

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

שליחת משימה לקריאה ולכתיבה של טבלאות Hudi

אחרי שיוצרים אשכול עם רכיב Hudi, אפשר לשלוח משימות Spark ו-Hive שקוראות וכותבות טבלאות Hudi.

דוגמה ל-gcloud CLI:

gcloud dataproc jobs submit pyspark \
    --cluster=CLUSTER_NAME \
    --region=region \
    JOB_FILE \
    -- JOB_ARGS

משימת PySpark לדוגמה

קובץ ה-PySpark הבא יוצר, קורא וכותב טבלת Hudi.

#!/usr/bin/env python
"""Pyspark Hudi test."""

import sys
from pyspark.sql import SparkSession


def create_hudi_table(spark, table_name, table_uri):
  """Creates Hudi table."""
  create_table_sql = f"""
    CREATE TABLE IF NOT EXISTS {table_name} (
      uuid string,
      begin_lat double,
      begin_lon double,
      end_lat double,
      end_lon double,
      driver string,
      rider string,
      fare double,
      partitionpath string,
      ts long
    ) USING hudi
    LOCATION '{table_uri}'
    TBLPROPERTIES (
      type = 'cow',
      primaryKey = 'uuid',
      preCombineField = 'ts'
    )
    PARTITIONED BY (partitionpath)
  """
  spark.sql(create_table_sql)


def generate_test_dataframe(spark, n_rows):
  """Generates test dataframe with Hudi's built-in data generator."""
  sc = spark.sparkContext
  utils = sc._jvm.org.apache.hudi.QuickstartUtils
  data_generator = utils.DataGenerator()
  inserts = utils.convertToStringList(data_generator.generateInserts(n_rows))
  return spark.read.json(sc.parallelize(inserts, 2))


def write_hudi_table(table_name, table_uri, df):
  """Writes Hudi table."""
  hudi_options = {
      'hoodie.table.name': table_name,
      'hoodie.datasource.write.recordkey.field': 'uuid',
      'hoodie.datasource.write.partitionpath.field': 'partitionpath',
      'hoodie.datasource.write.table.name': table_name,
      'hoodie.datasource.write.operation': 'upsert',
      'hoodie.datasource.write.precombine.field': 'ts',
      'hoodie.upsert.shuffle.parallelism': 2,
      'hoodie.insert.shuffle.parallelism': 2,
  }
  df.write.format('hudi').options(**hudi_options).mode('append').save(table_uri)


def query_commit_history(spark, table_name, table_uri):
  tmp_table = f'{table_name}_commit_history'
  spark.read.format('hudi').load(table_uri).createOrReplaceTempView(tmp_table)
  query = f"""
    SELECT DISTINCT(_hoodie_commit_time)
    FROM {tmp_table}
    ORDER BY _hoodie_commit_time
    DESC
  """
  return spark.sql(query)


def read_hudi_table(spark, table_name, table_uri, commit_ts=''):
  """Reads Hudi table at the given commit timestamp."""
  if commit_ts:
    options = {'as.of.instant': commit_ts}
  else:
    options = {}
  tmp_table = f'{table_name}_snapshot'
  spark.read.format('hudi').options(**options).load(
      table_uri
  ).createOrReplaceTempView(tmp_table)
  query = f"""
    SELECT _hoodie_commit_time, begin_lat, begin_lon,
        driver, end_lat, end_lon, fare, partitionpath,
        rider, ts, uuid
    FROM {tmp_table}
  """
  return spark.sql(query)


def main():
  """Test create write and read Hudi table."""
  if len(sys.argv) != 3:
    raise Exception('Expected arguments: <table_name> <table_uri>')

  table_name = sys.argv[1]
  table_uri = sys.argv[2]

  app_name = f'pyspark-hudi-test_{table_name}'
  print(f'Creating Spark session {app_name} ...')
  spark = SparkSession.builder.appName(app_name).getOrCreate()
  spark.sparkContext.setLogLevel('WARN')

  print(f'Creating Hudi table {table_name} at {table_uri} ...')
  create_hudi_table(spark, table_name, table_uri)

  print('Generating test data batch 1...')
  n_rows1 = 10
  input_df1 = generate_test_dataframe(spark, n_rows1)
  input_df1.show(truncate=False)

  print('Writing Hudi table, batch 1 ...')
  write_hudi_table(table_name, table_uri, input_df1)

  print('Generating test data batch 2...')
  n_rows2 = 10
  input_df2 = generate_test_dataframe(spark, n_rows2)
  input_df2.show(truncate=False)

  print('Writing Hudi table, batch 2 ...')
  write_hudi_table(table_name, table_uri, input_df2)

  print('Querying commit history ...')
  commits_df = query_commit_history(spark, table_name, table_uri)
  commits_df.show(truncate=False)
  previous_commit_ts = commits_df.collect()[1]._hoodie_commit_time

  print('Reading the Hudi table snapshot at the latest commit ...')
  output_df1 = read_hudi_table(spark, table_name, table_uri)
  output_df1.show(truncate=False)

  print(f'Reading the Hudi table snapshot at {previous_commit_ts} ...')
  output_df2 = read_hudi_table(spark, table_name, table_uri, previous_commit_ts)
  output_df2.show(truncate=False)

  print('Stopping Spark session ...')
  spark.stop()

  print('All done')


main()

הפקודה הבאה ב-CLI של gcloud שולחת את קובץ ה-PySpark לדוגמה אל Managed Service for Apache Spark.

gcloud dataproc jobs submit pyspark \
    --cluster=CLUSTER_NAME \
    gs://BUCKET_NAME/pyspark_hudi_example.py \
    -- TABLE_NAME gs://BUCKET_NAME/TABLE_NAME

שימוש ב-CLI של Hudi

‫Hudi CLI נמצא ב-/usr/lib/hudi/cli/hudi-cli.sh בצומת הראשי של אשכול Managed Service for Apache Spark. אתם יכולים להשתמש ב-Hudi CLI כדי להציג סכימות, נתוני קומיט ונתונים סטטיסטיים של טבלאות Hudi, ולבצע באופן ידני פעולות ניהוליות, כמו תזמון דחיסות (ראו שימוש ב-hudi-cli).

כדי להפעיל את ה-CLI של Hudi ולהתחבר לטבלת Hudi:

1. מבצעים SSH לצומת הראשי.
2. מריצים את /usr/lib/hudi/cli/hudi-cli.sh. שורת הפקודה משתנה ל-hudi->.
3. מריצים את connect --path gs://my-bucket/my-hudi-table.
4. מריצים פקודות, כמו desc, שמתארת את סכימת הטבלה, או commits show, שמציגה את היסטוריית ה-commit.
5. כדי להפסיק את הסשן של ה-CLI, מריצים את הפקודה exit.

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

• למדריך למתחילים של Hudi