使用 Spark

[hoodie]$ spark-submit \
  --packages org.apache.hudi:hudi-utilities-slim-bundle_2.12:1.2.0,org.apache.hudi:hudi-spark3.5-bundle_2.12:1.2.0 \
  --class org.apache.hudi.utilities.streamer.HoodieStreamer `ls packaging/hudi-utilities-slim-bundle/target/hudi-utilities-slim-bundle-*.jar` --help
Usage: <main class> [options]
  Options:
    --allow-commit-on-no-checkpoint-change
      allow commits even if checkpoint has not changed before and after fetch
      datafrom source. This might be useful in sources like SqlSource where
      there is not checkpoint. And is not recommended to enable in continuous
      mode.
      Default: false
    --base-file-format
      File format for the base files. PARQUET (or) HFILE
      Default: PARQUET
    --bootstrap-index-class
      subclass of BootstrapIndex
      Default: org.apache.hudi.common.bootstrap.index.HFileBootstrapIndex
    --bootstrap-overwrite
      Overwrite existing target table, default false
      Default: false
    --checkpoint
      Resume Hudi Streamer from this checkpoint.
    --cluster-scheduling-minshare
      Minshare for clustering as defined in
      https://spark.apache.org/docs/latest/job-scheduling.html
      Default: 0
    --cluster-scheduling-weight
      Scheduling weight for clustering as defined in
      https://spark.apache.org/docs/latest/job-scheduling.html
      Default: 1
    --commit-on-errors
      Commit even when some records failed to be written
      Default: false
    --compact-scheduling-minshare
      Minshare for compaction as defined in
      https://spark.apache.org/docs/latest/job-scheduling.html
      Default: 0
    --compact-scheduling-weight
      Scheduling weight for compaction as defined in
      https://spark.apache.org/docs/latest/job-scheduling.html
      Default: 1
    --config-hot-update-strategy-class
      Configuration hot update in continuous mode
      Default: <empty string>
    --continuous
      Hudi Streamer runs in continuous mode running source-fetch -> Transform
      -> Hudi Write in loop
      Default: false
    --delta-sync-scheduling-minshare
      Minshare for delta sync as defined in
      https://spark.apache.org/docs/latest/job-scheduling.html
      Default: 0
    --delta-sync-scheduling-weight
      Scheduling weight for delta sync as defined in
      https://spark.apache.org/docs/latest/job-scheduling.html
      Default: 1
    --disable-compaction
      Compaction is enabled for MoR table by default. This flag disables it
      Default: false
    --enable-hive-sync
      Enable syncing to hive
      Default: false
    --enable-sync
      Enable syncing meta
      Default: false
    --filter-dupes
      Should duplicate records from source be dropped/filtered out before
      insert/bulk-insert
      Default: false
    --force-empty-sync
      Force syncing meta even on empty commit
      Default: false
    --help, -h

    --hoodie-conf
      Any configuration that can be set in the properties file (using the CLI
      parameter "--props") can also be passed command line using this
      parameter. This can be repeated
      Default: []
    --ingestion-metrics-class
      Ingestion metrics class for reporting metrics during ingestion
      lifecycles.
      Default: org.apache.hudi.utilities.streamer.HoodieStreamerMetrics
    --initial-checkpoint-provider
      subclass of
      org.apache.hudi.utilities.checkpointing.InitialCheckpointProvider.
      Generate check point for Hudi Streamer for the first run. This field
      will override the checkpoint of last commit using the checkpoint field.
      Use this field only when switching source, for example, from DFS source
      to Kafka Source.
    --max-pending-clustering
      Maximum number of outstanding inflight/requested clustering. Delta Sync
      will not happen unlessoutstanding clustering is less than this number
      Default: 5
    --max-pending-compactions
      Maximum number of outstanding inflight/requested compactions. Delta Sync
      will not happen unlessoutstanding compactions is less than this number
      Default: 5
    --max-retry-count
      the max retry count if --retry-on-source-failures is enabled
      Default: 3
    --min-sync-interval-seconds
      the min sync interval of each sync in continuous mode
      Default: 0
    --op
      Takes one of these values : UPSERT (default), INSERT, BULK_INSERT,
      INSERT_OVERWRITE, INSERT_OVERWRITE_TABLE, DELETE_PARTITION, DELETE
      (DELETE extracts HoodieKeys from source records and deletes the
      corresponding records from the table.)
      Default: UPSERT
      Possible Values: [INSERT, INSERT_PREPPED, UPSERT, UPSERT_PREPPED, BULK_INSERT, BULK_INSERT_PREPPED, DELETE, DELETE_PREPPED, BOOTSTRAP, INSERT_OVERWRITE, CLUSTER, DELETE_PARTITION, INSERT_OVERWRITE_TABLE, COMPACT, INDEX, ALTER_SCHEMA, LOG_COMPACT, UNKNOWN]
    --payload-class
      subclass of HoodieRecordPayload, that works off a GenericRecord.
      Implement your own, if you want to do something other than overwriting
      existing value
      Default: org.apache.hudi.common.model.OverwriteWithLatestAvroPayload
    --post-write-termination-strategy-class
      Post writer termination strategy class to gracefully shutdown
      Hudi Streamer in continuous mode
      Default: <empty string>
    --props
      path to properties file on localfs or dfs, with configurations for
      hoodie client, schema provider, key generator and data source. For
      hoodie client props, sane defaults are used, but recommend use to
      provide basic things like metrics endpoints, hive configs etc. For
      sources, referto individual classes, for supported properties.
      Properties in this file can be overridden by "--hoodie-conf"
      Default: file:///Users/shiyanxu/src/test/resources/streamer-config/dfs-source.properties
    --retry-interval-seconds
      the retry interval for source failures if --retry-on-source-failures is
      enabled
      Default: 30
    --retry-last-pending-inline-clustering, -rc
      Retry last pending inline clustering plan before writing to sink.
      Default: false
    --retry-last-pending-inline-compaction
      Retry last pending inline compaction plan before writing to sink.
      Default: false
    --retry-on-source-failures
      Retry on any source failures
      Default: false
    --run-bootstrap
      Run bootstrap if bootstrap index is not found
      Default: false
    --schemaprovider-class
      subclass of org.apache.hudi.utilities.schema.SchemaProvider to attach
      schemas to input & target table data, built in options:
      org.apache.hudi.utilities.schema.FilebasedSchemaProvider.Source (See
      org.apache.hudi.utilities.sources.Source) implementation can implement
      their own SchemaProvider. For Sources that return Dataset<Row>, the
      schema is obtained implicitly. However, this CLI option allows
      overriding the schemaprovider returned by Source.
    --source-class
      Subclass of org.apache.hudi.utilities.sources to read data. Built-in
      options: org.apache.hudi.utilities.sources.{JsonDFSSource (default),
      AvroDFSSource, JsonKafkaSource, AvroKafkaSource, HiveIncrPullSource}
      Default: org.apache.hudi.utilities.sources.JsonDFSSource
    --source-limit
      Maximum amount of data to read from source. Default: No limit, e.g:
      DFS-Source => max bytes to read, Kafka-Source => max events to read
      Default: 9223372036854775807
    --source-ordering-field
      Field within source record to decide how to break ties between records
      with same key in input data. Default: 'ts' holding unix timestamp of
      record
      Default: ts
    --spark-master
      spark master to use, if not defined inherits from your environment
      taking into account Spark Configuration priority rules (e.g. not using
      spark-submit command).
      Default: <empty string>
    --sync-tool-classes
      Meta sync client tool, using comma to separate multi tools
      Default: org.apache.hudi.hive.HiveSyncTool
  * --table-type
      Type of table. COPY_ON_WRITE (or) MERGE_ON_READ
  * --target-base-path
      base path for the target hoodie table. (Will be created if did not exist
      first time around. If exists, expected to be a hoodie table)
  * --target-table
      name of the target table
    --transformer-class
      A subclass or a list of subclasses of
      org.apache.hudi.utilities.transform.Transformer. Allows transforming raw
      source Dataset to a target Dataset (conforming to target schema) before
      writing. Default : Not set. E.g. -
      org.apache.hudi.utilities.transform.SqlQueryBasedTransformer (which
      allows a SQL query templated to be passed as a transformation function).
      Pass a comma-separated list of subclass names to chain the
      transformations. If there are two or more transformers using the same
      config keys and expect different values for those keys, then transformer
      can include an identifier. E.g. -
      tr1:org.apache.hudi.utilities.transform.SqlQueryBasedTransformer. Here
      the identifier tr1 can be used along with property key like
      `hoodie.streamer.transformer.sql.tr1` to identify properties related to
      the transformer. So effective value for
      `hoodie.streamer.transformer.sql` is determined by key
      `hoodie.streamer.transformer.sql.tr1` for this transformer. If
      identifier is used, it should be specified for all the transformers.
      Further the order in which transformer is applied is determined by the
      occurrence of transformer irrespective of the identifier used for the
      transformer. For example: In the configured value below tr2:org.apache.hudi.utilities.transform.SqlQueryBasedTransformer,tr1:org.apache.hudi.utilities.transform.SqlQueryBasedTransformer
      , tr2 is applied before tr1 based on order of occurrence.

[confluent-5.0.0]$ bin/ksql-datagen schema=../impressions.avro format=avro topic=impressions key=impressionid

[hoodie]$ spark-submit \
  --packages org.apache.hudi:hudi-utilities-slim-bundle_2.12:1.2.0,org.apache.hudi:hudi-spark3.5-bundle_2.12:1.2.0 \
  --class org.apache.hudi.utilities.streamer.HoodieStreamer `ls packaging/hudi-utilities-slim-bundle/target/hudi-utilities-slim-bundle-*.jar` \
  --props file://${PWD}/hudi-utilities/src/test/resources/streamer-config/kafka-source.properties \
  --schemaprovider-class org.apache.hudi.utilities.schema.SchemaRegistryProvider \
  --source-class org.apache.hudi.utilities.sources.AvroKafkaSource \
  --source-ordering-field impresssiontime \
  --target-base-path file:\/\/\/tmp/hudi-streamer-op \
  --target-table uber.impressions \
  --op BULK_INSERT

hoodie.write.concurrency.mode=optimistic_concurrency_control
hoodie.write.lock.provider=<lock-provider-classname>
hoodie.cleaner.policy.failed.writes=LAZY

[hoodie]$ spark-submit \
  --packages org.apache.hudi:hudi-utilities-slim-bundle_2.12:1.2.0,org.apache.hudi:hudi-spark3.5-bundle_2.12:1.2.0 \
  --class org.apache.hudi.utilities.streamer.HoodieStreamer `ls packaging/hudi-utilities-slim-bundle/target/hudi-utilities-slim-bundle-*.jar` \
  --props file://${PWD}/hudi-utilities/src/test/resources/streamer-config/kafka-source.properties \
  --schemaprovider-class org.apache.hudi.utilities.schema.SchemaRegistryProvider \
  --source-class org.apache.hudi.utilities.sources.AvroKafkaSource \
  --source-ordering-field impresssiontime \
  --target-base-path file:///tmp/hudi-streamer-op \
  --target-table uber.impressions \
  --op BULK_INSERT

# resume after the instant with this requested instant time
--checkpoint resumeFromInstantRequestTime:20250110120000000

# resume after the instant with this completion time
--checkpoint resumeFromInstantCompletionTime:20250110120005000

Illegal checkpoint key override `20250110120000000`. Valid format is either
`resumeFromInstantRequestTime:<checkpoint value>` or `resumeFromInstantCompletionTime:<checkpoint value>`.

--transformer-class org.apache.hudi.utilities.transform.SqlQueryBasedTransformer
--hoodie-conf hoodie.streamer.transformer.sql=SELECT a.col1, a.col3, a.col4 FROM <SRC> a

CACHE TABLE tmp_personal_trips AS
SELECT * FROM <SRC> WHERE trip_type='personal_trips';

SELECT * FROM tmp_personal_trips;

age as intColumn,address as stringColumn,name.first as name_first,name.last as name_last, name.middle as name_middle

--transformer-class org.apache.hudi.utilities.transform.FlatteningTransformer

--transformer-class org.apache.hudi.utilities.transform.FlatteningTransformer,org.apache.hudi.utilities.transform.SqlQueryBasedTransformer
--hoodie-conf hoodie.streamer.transformer.sql=SELECT a.col1, a.col3, a.col4 FROM <SRC> a

--transformer-class org.apache.hudi.utilities.transform.AWSDmsTransformer

--hoodie-conf hoodie.streamer.schemaprovider.registry.url=https://foo:bar@schemaregistry.org

streamName,shardId:value,shardId:value,...

my-stream,shardId-000000000000:49590…88898,shardId-000000000001:49590…96306

# kinesis-source.properties
hoodie.streamer.source.kinesis.stream.name=my-stream
hoodie.streamer.source.kinesis.region=us-east-1
hoodie.streamer.source.kinesis.starting.position=LATEST

# Standard Hudi write / key-gen configs
hoodie.datasource.write.recordkey.field=id
hoodie.datasource.write.partitionpath.field=event_date
hoodie.table.ordering.fields=ts

spark-submit \
  --packages org.apache.hudi:hudi-utilities-slim-bundle_2.12:1.2.0,org.apache.hudi:hudi-spark3.5-bundle_2.12:1.2.0 \
  --class org.apache.hudi.utilities.streamer.HoodieStreamer \
  hudi-utilities-slim-bundle-*.jar \
  --props kinesis-source.properties \
  --source-class org.apache.hudi.utilities.sources.JsonKinesisSource \
  --table-type COPY_ON_WRITE \
  --target-base-path s3://my-bucket/hudi/my-table \
  --target-table my_db.my_table \
  --op UPSERT \
  --continuous

hoodie.streamer.source.kafka.topic=postgres.public.customers
hoodie.streamer.schemaprovider.registry.url=http://localhost:8081/subjects/postgres.public.customers-value/versions/latest
bootstrap.servers=localhost:9092
auto.offset.reset=earliest
schema.registry.url=http://localhost:8081

hoodie.datasource.write.recordkey.field=id

[hoodie]$ spark-submit \
  --packages org.apache.hudi:hudi-utilities-slim-bundle_2.12:1.2.0,org.apache.hudi:hudi-spark3.5-bundle_2.12:1.2.0 \
  --class org.apache.hudi.utilities.streamer.HoodieStreamer `ls packaging/hudi-utilities-slim-bundle/target/hudi-utilities-slim-bundle-*.jar` \
  --props file://${PWD}/debezium-source.properties \
  --schemaprovider-class org.apache.hudi.utilities.schema.SchemaRegistryProvider \
  --source-class org.apache.hudi.utilities.sources.debezium.PostgresDebeziumSource \
  --payload-class org.apache.hudi.common.model.debezium.PostgresDebeziumAvroPayload \
  --source-ordering-field _event_lsn \
  --target-base-path file:///tmp/hudi-debezium-customers \
  --target-table customers \
  --table-type MERGE_ON_READ \
  --op UPSERT \
  --continuous

  --source-class org.apache.hudi.utilities.sources.debezium.MysqlDebeziumSource \
  --payload-class org.apache.hudi.common.model.debezium.MySqlDebeziumAvroPayload \
  --source-ordering-field _event_seq \

/**
 * Post write termination strategy for Hudi Streamer in continuous mode.
 */
public interface PostWriteTerminationStrategy {

  /**
   * Returns whether HoodieStreamer needs to be shutdown.
   * @param scheduledCompactionInstantAndWriteStatuses optional pair of scheduled compaction instant and write statuses.
   * @return true if HoodieStreamer has to be shutdown. false otherwise.
   */
  boolean shouldShutdown(Option<Pair<Option<String>, JavaRDD<WriteStatus>>> scheduledCompactionInstantAndWriteStatuses);

}

  * --config-folder
    the path to the folder which contains all the table wise config files
    --base-path-prefix
    this is added to enable users to create all the hudi datasets for related tables under one path in FS. The datasets are then created under the path - <base_path_prefix>/<database>/<table_to_be_ingested>. However you can override the paths for every table by setting the property hoodie.streamer.ingestion.targetBasePath

hoodie.streamer.ingestion.tablesToBeIngested
  comma separated names of tables to be ingested in the format <database>.<table>, for example db1.table1,db1.table2
hoodie.streamer.ingestion.targetBasePath
  if you wish to ingest a particular table in a separate path, you can mention that path here
hoodie.streamer.ingestion.<database>.<table>.configFile
  path to the config file in dedicated config folder which contains table overridden properties for the particular table to be ingested.

[hoodie]$ spark-submit \
  --packages org.apache.hudi:hudi-utilities-slim-bundle_2.12:1.2.0,org.apache.hudi:hudi-spark3.5-bundle_2.12:1.2.0 \
  --class org.apache.hudi.utilities.streamer.HoodieMultiTableStreamer `ls packaging/hudi-utilities-slim-bundle/target/hudi-utilities-slim-bundle-*.jar` \
  --props file://${PWD}/hudi-utilities/src/test/resources/streamer-config/kafka-source.properties \
  --config-folder file://tmp/hudi-ingestion-config \
  --schemaprovider-class org.apache.hudi.utilities.schema.SchemaRegistryProvider \
  --source-class org.apache.hudi.utilities.sources.AvroKafkaSource \
  --source-ordering-field impresssiontime \
  --base-path-prefix file:\/\/\/tmp/hudi-streamer-op \
  --target-table uber.impressions \
  --op BULK_INSERT

spark-submit \
  --packages org.apache.hudi:hudi-utilities-slim-bundle_2.12:1.2.0,org.apache.hudi:hudi-spark3.5-bundle_2.12:1.2.0 \
  --class org.apache.hudi.utilities.HudiHiveSyncJob \
  hudi-utilities-slim-bundle-*.jar \
  --base-path s3://my-bucket/hudi/my-table \
  --base-file-format PARQUET \
  --hoodie-conf hoodie.datasource.hive_sync.mode=hms \
  --hoodie-conf hoodie.datasource.hive_sync.metastore.uris=thrift://hive-metastore:9083 \
  --hoodie-conf hoodie.datasource.hive_sync.database=my_db \
  --hoodie-conf hoodie.datasource.hive_sync.table=my_table