רכיב Hudi אופציונלי ב-Dataproc

אפשר להתקין רכיבים נוספים כמו Hudi כשיוצרים אשכול Dataproc באמצעות התכונה רכיבים אופציונליים. בדף הזה מוסבר איך להתקין את רכיב Hudi באשכול Dataproc.

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

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

אפשר להתקין את רכיב Hudi באשכולות Dataproc שנוצרו עם הגרסאות הבאות של תמונות Dataproc:

• 2.0.54+
• 2.1.2+
• 2.2.31+

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

כשיוצרים אשכול Dataproc עם 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 כשיוצרים אשכול Dataproc.

בדפים של גרסאות הפצה של Dataproc Image מפורטת גרסת רכיב Hudi שכלולה בכל גרסת הפצה של Dataproc Image.

המסוף

1. מפעילים את הרכיב.
   • במסוף Google Cloud , פותחים את הדף Dataproc Create a cluster. החלונית הגדרת אשכול נבחרת.
   • בקטע Components (רכיבים):
     • בקטע Optional components (רכיבים אופציונליים), בוחרים ברכיב Hudi.

פקודת gcloud

כדי ליצור אשכול Dataproc שכולל את רכיב 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: אופציונלי. אפשר להשתמש בדגל האופציונלי הזה כדי לציין גרסת תמונה של Dataproc שאינה ברירת המחדל (ראו גרסת ברירת המחדל של תמונת Dataproc).
• ‫PROPERTIES: אופציונלי. אפשר להשתמש בדגל האופציונלי הזה כדי להגדיר מאפיינים של רכיב Hudi, שמצוינים באמצעות hudi: קידומת הקובץ (לדוגמה: properties=hudi:hoodie.datasource.write.table.type=COPY_ON_WRITE).
  • מאפיין גרסת רכיב Hudi: אפשר לציין את המאפיין dataproc:hudi.version. הערה: גרסת רכיב Hudi מוגדרת על ידי Dataproc כך שתהיה תואמת לגרסת תמונת אשכול Dataproc. אם מגדירים את המאפיין הזה, יכול להיות שיצירת האשכול תיכשל אם הגרסה שצוינה לא תואמת לתמונת האשכול.
  • מאפייני Spark ו-Hive: ‏ Dataproc מגדיר מאפיינים שקשורים ל-Hudi Spark ו-Hive כשיוצרים את האשכול. לא צריך להגדיר אותן כשיוצרים את האשכול או כששולחים עבודות.

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 לדוגמה אל Dataproc.

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

שימוש ב-Hudi CLI

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

כדי להפעיל את Hudi CLI ולהתחבר לטבלת 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 שמציגה את היסטוריית השמירה.
5. כדי להפסיק את סשן ה-CLI, מריצים את הפקודה exit.

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

• מדריך למתחילים בנושא Hudi