Componente opcional Hudi do Managed Service for Apache Spark

O componente Apache Hudi permite executar o Hudi no Serviço Gerenciado para Apache Spark. Quando você instala o componente durante a criação do cluster usando o recurso Componentes opcionais, ele implanta bibliotecas do Hudi e configura o Spark e o Hive no cluster para trabalhar com o Hudi.

Versões de imagem compatíveis do Serviço Gerenciado para Apache Spark

É possível instalar o componente Hudi em clusters do Serviço Gerenciado para Apache Spark criados com as seguintes versões de imagem do Serviço Gerenciado para Apache Spark:

Propriedades relacionadas ao Hudi

Quando você cria um cluster do Serviço Gerenciado para Apache Spark com o Hudi, as seguintes propriedades do Spark e do Hive são configuradas para funcionar com o Hudi.

Arquivo de configuração Propriedade Valor padrão
/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

Instalar o componente

Instale o componente Hudi ao criar um cluster do Serviço Gerenciado para Apache Spark.

As páginas de versão de lançamento da imagem do Serviço Gerenciado para Apache Spark listam a versão do componente Hudi incluída em cada versão de lançamento da imagem do Serviço Gerenciado para Apache Spark.

Console doGoogle Cloud

  1. No console do Google Cloud , abra a página Criar cluster.
  2. Clique em Configuração adicional para abrir a seção.
  3. Edite os Componentes opcionais.
  4. No painel que será aberto, marque a caixa de seleção de Hudi.
  5. Clique em Salvar.

CLI da gcloud

Para criar um cluster do Serviço Gerenciado para Apache Spark que inclua o componente Hudi, use o comando com a flag --optional-components.

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

Substitua:

  • CLUSTER_NAME: obrigatório. O novo nome do cluster.
  • REGION: obrigatório. A região do cluster.
  • DATAPROC_IMAGE: opcional. Use essa flag opcional para especificar uma versão de imagem do Serviço Gerenciado para Apache Spark que não seja padrão (consulte Versão de imagem padrão do Serviço Gerenciado para Apache Spark).
  • PROPERTIES: opcional. Use essa sinalização opcional para definir as propriedades do componente Hudi, que são especificadas com o prefixo de arquivo hudi:. Exemplo: properties=hudi:hoodie.datasource.write.table.type=COPY_ON_WRITE).
    • Propriedade da versão do componente Hudi: é possível especificar a propriedade dataproc:hudi.version. Observação: a versão do componente do Hudi é definida pelo Serviço Gerenciado para Apache Spark para ser compatível com a versão da imagem do cluster do Serviço Gerenciado para Apache Spark. Se você definir essa propriedade, a criação do cluster poderá falhar se a versão especificada não for compatível com a imagem do cluster.
    • Propriedades do Spark e do Hive: o Serviço Gerenciado para Apache Spark define propriedades do Spark e do Hive relacionadas ao Hudi quando o cluster é criado. Não é necessário defini-los ao criar o cluster ou enviar jobs.

API REST

O componente Hudi pode ser instalado pela API Dataproc usando SoftwareConfig.Component como parte de uma solicitação clusters.create.

Enviar um job para ler e gravar tabelas do Hudi

Depois de criar um cluster com o componente Hudi, envie jobs do Spark e do Hive que leem e gravam tabelas do Hudi.

Exemplo de gcloud CLI:

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

Exemplo de job do PySpark

O arquivo PySpark a seguir cria, lê e grava uma tabela do 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()

O comando da CLI gcloud a seguir envia o arquivo de amostra do PySpark para o Serviço Gerenciado para Apache Spark.

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

Usar a CLI do Hudi

A CLI do Hudi está localizada em /usr/lib/hudi/cli/hudi-cli.sh no nó mestre do cluster do Serviço Gerenciado para Apache Spark. É possível usar a CLI do Hudi para visualizar esquemas, confirmações e estatísticas de tabelas do Hudi e realizar manualmente operações administrativas, como programar compactações (consulte Como usar o hudi-cli).

Para iniciar a CLI do Hudi e se conectar a uma tabela do Hudi:

  1. Conecte-se por SSH ao nó mestre.
  2. Execute /usr/lib/hudi/cli/hudi-cli.sh. O prompt de comando muda para hudi->.
  3. Execute connect --path gs://my-bucket/my-hudi-table.
  4. Execute comandos, como desc, que descreve o esquema da tabela, ou commits show, que mostra o histórico de commits.
  5. Para interromper a sessão da CLI, execute exit.

A seguir