Optionale Hudi-Komponente von Dataproc

Sie können zusätzliche Komponenten wie Hudi installieren, wenn Sie einen Dataproc-Cluster mit der Funktion Optionale Komponenten erstellen. Auf dieser Seite wird beschrieben, wie Sie optional die Hudi-Komponente in einem Dataproc-Cluster installieren können.

Bei Installation in einem Dataproc-Cluster installiert die Apache Hudi-Komponente Hudi-Bibliotheken und konfiguriert Spark und Hive im Cluster für Hudi.

Kompatible Dataproc-Image-Versionen

Sie können die Hudi-Komponente auf Dataproc-Clustern installieren, die mit den folgenden Dataproc-Image-Versionen erstellt wurden:

• 1.5.80+
• 2.0.54+
• 2.1.2+

Hudi-Eigenschaften

Wenn Sie einen Dataproc mit Hudi-Cluster erstellen, werden die folgenden Spark- und Hive-Attribute für die Zusammenarbeit mit Hudi konfiguriert.

Konfigurationsdatei	Attribut	Standardwert
/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

Komponente installieren

Hudi-Komponente installieren, wenn Sie einen Dataproc-Cluster erstellen

Auf den Seiten der Dataproc-Image-Versionen wird die in jedem Dataproc-Image-Release enthaltene Hudi-Komponentenversion aufgeführt.

Console

1. Aktivieren Sie die Komponente.
   • Rufen Sie in der Google Cloud Console die Dataproc-Seite Cluster erstellen auf. Der Bereich Cluster einrichten ist ausgewählt.
   • Im Bereich Komponenten:
     • Wählen Sie unter Optionale Komponenten die Komponente Hudi aus.

gcloud-Befehl

Verwenden Sie den Befehl mit dem Flag --optional-components, um einen Dataproc-Cluster zu erstellen, der die Hudi-Komponente enthält.

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

Ersetzen Sie Folgendes:

• CLUSTER_NAME: erforderlich. Der Name des neuen Clusters.
• REGION: erforderlich. Die Clusterregion.
• DATAPROC_IMAGE: Optional. Sie können dieses optionale Flag verwenden, um eine nicht standardmäßige Dataproc-Image-Version anzugeben (siehe Dataproc-Standard-Image-Version).
• PROPERTIES: Optional. Mit diesem optionalen Flag können Sie die Eigenschaften der Hudi-Komponente festlegen, die mit dem Dateipräfix hudi: angegeben werden. Beispiel: properties=hudi:hoodie.datasource.write.table.type=COPY_ON_WRITE.
  • Eigenschaft der Hudi-Komponentenversion: Optional kannst du die Eigenschaft dataproc:hudi.version angeben. Hinweis:Die Version der Hudi-Komponente wird von Dataproc so festgelegt, dass sie mit der Image-Version des Dataproc-Clusters kompatibel ist. Wenn Sie dieses Attribut festlegen, kann die Clustererstellung fehlschlagen, wenn die angegebene Version nicht mit dem Cluster-Image kompatibel ist.
  • Spark- und Hive-Attribute: Dataproc legt beim Erstellen des Clusters Hudi-bezogene Spark- und Hive-Attribute fest. Sie müssen sie beim Erstellen des Clusters oder Senden von Jobs nicht festlegen.

REST API

Die Hudi-Komponente kann über die Dataproc API mit SoftwareConfig.Component im Rahmen einer clusters.create-Anfrage installiert werden.

Job zum Lesen und Schreiben von Hudi-Tabellen senden

Nach dem Erstellen eines Clusters mit der Hudi-Komponente können Sie Spark- und Hive-Jobs senden, die Hudi-Tabellen lesen und schreiben.

gcloud CLI-Beispiel:

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

PySpark-Beispieljob

Die folgende PySpark-Datei erstellt, liest und schreibt eine Hudi-Tabelle.

#!/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()

Mit dem folgenden gcloud CLI-Befehl wird die PySpark-Beispieldatei an Dataproc gesendet.

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

Hudi CLI verwenden

Die Hudi-Befehlszeile befindet sich unter /usr/lib/hudi/cli/hudi-cli.sh auf dem Masterknoten des Dataproc-Clusters. Sie können die Hudi CLI verwenden, um Hudi-Tabellenschemas, Commits und Statistiken anzusehen und Verwaltungsvorgänge wie geplante Verdichtungen manuell auszuführen (siehe hudi-cli).

So starten Sie die Hudi CLI und stellen eine Verbindung zu einem Hudi-Tisch her:

1. Stellen Sie eine SSH-Verbindung zum Master-Knoten her.
2. Führen Sie /usr/lib/hudi/cli/hudi-cli.sh aus. Die Eingabeaufforderung ändert sich in hudi->.
3. Führen Sie connect --path gs://my-bucket/my-hudi-table aus.
4. Führen Sie Befehle aus, z. B. desc zum Beschreiben des Tabellenschemas oder commits show zum Anzeigen des Commit-Verlaufs.
5. Führen Sie exit aus, um die Befehlszeilensitzung zu beenden.

Weitere Informationen

• Hudi-Kurzanleitung