Skip to main content
Version: Next

Hudi

Hudi sink connector

Description​

Used to write data to Hudi.

Key features​

Hive Metastore synchronization

The SeaTunnel Hudi sink writes Hudi data files and .hoodie metadata, but it does not register the table in or synchronize the table with Hive Metastore. Options matching hoodie.datasource.hive_sync.* are not supported sink options and are not passed to the Hudi write client. Run Apache Hudi's HiveSyncTool or another table registration process separately when Hive Metastore registration is required.

Options​

Base configuration:

nametyperequireddefault valuedescription
table_dfs_pathstringyes-Root path for Hudi table data and metadata.
conf_files_pathstringno-Semicolon-separated local paths to HDFS client configuration files.
table_listArrayno-Per-table settings for multi-table jobs.
schema_save_modeenumnoCREATE_SCHEMA_WHEN_NOT_EXISTHow to handle the target schema before the job starts.
data_save_modeenumnoAPPEND_DATAHow to handle existing table data before the job starts.
common-optionsConfigno-Common sink options.

Table list configuration:

nametyperequireddefault valuedescription
table_namestringyes-Target Hudi table name.
databasestringnodefaultTarget Hudi database name.
table_typeenumnoCOPY_ON_WRITEHudi table type: COPY_ON_WRITE or MERGE_ON_READ.
op_typeenumnoINSERTWrite operation: INSERT, UPSERT, or BULK_INSERT.
record_key_fieldsstringno-Field or fields used to build the Hudi record key; required for UPSERT.
partition_fieldsstringno-Field or fields used to build the partition path.
precombine_fieldstringno-Field used to resolve multiple updates to the same record.
batch_interval_msIntno1000Currently unused; flushes are triggered only by batch_size or a checkpoint.
batch_sizeIntno1000Maximum buffered rows before a flush.
insert_shuffle_parallelismIntno2Shuffle parallelism for insert operations.
upsert_shuffle_parallelismIntno2Shuffle parallelism for upsert operations.
min_commits_to_keepIntno20Minimum number of commits retained during cleaning.
max_commits_to_keepIntno30Maximum number of commits retained during cleaning.
index_typeenumnoBLOOMHudi index type: BLOOM, SIMPLE, or GLOBAL_BLOOM.
index_class_namestringno-Fully qualified custom Hudi index class name.
record_byte_sizeIntno1024Estimated average record size in bytes.
cdc_enabledbooleannofalsePersist Hudi CDC change-log data when enabled.

Note: When this configuration corresponds to a single table, you can flatten the configuration items in table_list to the outer layer. For a multi-table job, keep table-specific options inside each table_list entry; table_dfs_path, conf_files_path, schema_save_mode, and data_save_mode remain at the sink level.

record_key_fields is required for UPSERT (validated at startup) and for BULK_INSERT (currently unvalidated — omitting it causes a NullPointerException during writing, not a config-time error). For CDC input, the upstream record must contain the fields used by record_key_fields; set cdc_enabled = true only when the Hudi CDC change log is required.

table_name [string]​

table_name The name of hudi table.

database [string]​

database The database of hudi table.

table_dfs_path [string]​

table_dfs_path The dfs root path of hudi table, such as 'hdfs://nameserivce/data/hudi/'.

table_type [enum]​

table_type The type of hudi table. The value is COPY_ON_WRITE or MERGE_ON_READ.

record_key_fields [string]​

record_key_fields The record key fields of hudi table, its are used to generate record key. It must be configured when op_type is UPSERT.

partition_fields [string]​

partition_fields The partition key fields of hudi table, its are used to generate partition.

precombine_field [string]​

precombine_field The precombine field of hudi table, its are used in preCombining before actual write.

index_type [string]​

index_type The index type of hudi table. Currently, BLOOM, SIMPLE, and GLOBAL_BLOOM are supported.

index_class_name [string]​

index_class_name The customized index classpath of hudi table, example org.apache.seatunnel.connectors.seatunnel.hudi.index.CustomHudiIndex.

record_byte_size [Int]​

record_byte_size The byte size of each record, This value can be used to help calculate the approximate number of records in each hudi data file. Adjusting this value can effectively reduce the number of hudi data file write magnifications.

conf_files_path [string]​

conf_files_path The environment conf file path list(local path), which used to init hdfs client to read hudi table file. The example is '/home/test/hdfs-site.xml;/home/test/core-site.xml;/home/test/yarn-site.xml'.

op_type [enum]​

op_type The operation type of hudi table. The value is insert or upsert or bulk_insert.

batch_interval_ms [Int]​

batch_interval_ms The maximum interval, in milliseconds, between two flushes to Hudi.

batch_size [Int]​

batch_size The maximum number of rows buffered before one flush to Hudi.

insert_shuffle_parallelism [Int]​

insert_shuffle_parallelism The parallelism of insert data to hudi table.

upsert_shuffle_parallelism [Int]​

upsert_shuffle_parallelism The parallelism of upsert data to hudi table.

min_commits_to_keep [Int]​

min_commits_to_keep The min commits to keep of hudi table.

max_commits_to_keep [Int]​

max_commits_to_keep The max commits to keep of hudi table.

cdc_enabled [boolean]​

cdc_enabled Whether to persist the CDC change log. When enable, persist the change data if necessary, and the table can be queried as a CDC query mode.

schema_save_mode [Enum]​

Controls how the connector handles the target table schema before the job starts:

  • RECREATE_SCHEMA: Create the table if it does not exist; otherwise delete and recreate it.
  • CREATE_SCHEMA_WHEN_NOT_EXIST: Create the table only when it does not exist.
  • ERROR_WHEN_SCHEMA_NOT_EXIST: Fail when the table does not exist.
  • IGNORE: Do not perform schema handling.

data_save_mode [Enum]​

Choose how to handle existing data before the synchronization task starts.

DROP_DATA: Keep the table structure and delete existing data.

APPEND_DATA: Keep the table structure and existing data.

ERROR_WHEN_DATA_EXISTS: Throw an error when data already exists.

common options​

Sink plugin common parameters, please refer to Sink Common Options for details.

Examples​

Single Table Upsert​

When op_type is UPSERT, record_key_fields must be configured.

sink {
  Hudi {
    table_dfs_path = "/tmp/seatunnel_mnt/hudi"
    database = "st"
    table_name = "st_test"
    table_type = "COPY_ON_WRITE"
    op_type = "UPSERT"
    record_key_fields = "c_bigint"
    batch_size = 1000
    batch_interval_ms = 1000
  }
}

Minimal Single Table​

For append-only writes, only table_dfs_path and table_name are required.

sink {
  Hudi {
    table_dfs_path = "/tmp/seatunnel_mnt/hudi"
    table_name = "st_test"
  }
}

Multiple Tables​

Use table_list when the upstream source produces multiple tables.

env {
  parallelism = 1
  job.mode = "STREAMING"
  checkpoint.interval = 5000
}

source {
  Mysql-CDC {
    url = "jdbc:mysql://127.0.0.1:3306/seatunnel"
    username = "root"
    password = "******"
    
    table-names = ["seatunnel.role","seatunnel.user","galileo.Bucket"]
  }
}

transform {
}

sink {
  Hudi {
    table_dfs_path = "hdfs://nameserivce/data/"
    conf_files_path = "/home/test/hdfs-site.xml;/home/test/core-site.xml;/home/test/yarn-site.xml"
    table_list = [
      {
        database = "st1"
        table_name = "role"
        table_type = "COPY_ON_WRITE"
        op_type = "INSERT"
        batch_size = 10000
      },
      {
        database = "st1"
        table_name = "user"
        table_type = "COPY_ON_WRITE"
        op_type = "UPSERT"
        record_key_fields = "user_id"
        batch_size = 10000
      },
      {
        database = "st1"
        table_name = "Bucket"
        table_type = "MERGE_ON_READ"
      }
    ]
  }
}

CDC To Hudi​

Enable cdc_enabled when the target Hudi table needs to persist CDC changelog information.

sink {
  Hudi {
    table_dfs_path = "/tmp/seatunnel_mnt/hudi"
    database = "st"
    table_name = "st_test"
    table_type = "COPY_ON_WRITE"
    op_type = "UPSERT"
    record_key_fields = "id"
    cdc_enabled = true
  }
}

S3 Storage​

The sink can write to an S3-compatible path. The connector-hudi module does not depend on hadoop-aws/aws-java-sdk, so resolving the s3a:// scheme requires placing hadoop-aws and the matching AWS SDK bundle (or SeaTunnel's seatunnel-hadoop-aws jar) into $SEATUNNEL_HOME/lib (or the connector's plugin lib directory) before the example below will work. After that, provide the required Hadoop filesystem settings through conf_files_path (or the runtime classpath), then use an s3a:// table path.

sink {
  Hudi {
    table_dfs_path = "s3a://hudi/"
    conf_files_path = "/etc/hadoop/core-site.xml;/etc/hadoop/hdfs-site.xml"
    table_name = "st_test"
    op_type = "UPSERT"
    record_key_fields = "id"
  }
}

Changelog​

Change Log
ChangeCommitVersion
[Fix][Core]fix kotlin jar conflict (#9683)https://github.com/apache/seatunnel/commit/c4ec5c0be52.3.12
[Improve][Connector-Hudi] Add pre-combine field option for hudi sink (#9496)https://github.com/apache/seatunnel/commit/f134d7e1292.3.12
[Feature][Checkpoint] Add check script for source/sink state class serialVersionUID missing (#9118)https://github.com/apache/seatunnel/commit/4f5adeb1c72.3.11
[improve] hudi options (#8952)https://github.com/apache/seatunnel/commit/b24d0e7f862.3.10
[Improve] restruct connector common options (#8634)https://github.com/apache/seatunnel/commit/f3499a6eeb2.3.10
[Improve][CI]skip ui module, improve module dependent (#8225)https://github.com/apache/seatunnel/commit/81de0a69cc2.3.9
[Feature][Connector-V2] Support write cdc changelog event into hudi sink (#7845)https://github.com/apache/seatunnel/commit/934434cc752.3.9
[Feature][Restapi] Allow metrics information to be associated to logical plan nodes (#7786)https://github.com/apache/seatunnel/commit/6b7c53d03c2.3.9
[Feature][Connector-V2] Optimize hudi sink (#7662)https://github.com/apache/seatunnel/commit/0d12520f912.3.8
[Improve][Connector] Add multi-table sink option check (#7360)https://github.com/apache/seatunnel/commit/2489f6446b2.3.7
[Feature][Core] Support using upstream table placeholders in sink options and auto replacement (#7131)https://github.com/apache/seatunnel/commit/c4ca74122c2.3.6
Bump org.xerial.snappy:snappy-java (#7144)https://github.com/apache/seatunnel/commit/aa26471fb72.3.6
[Feature][Connector-V2] [Hudi]Add hudi sink connector (#4405)https://github.com/apache/seatunnel/commit/dc271dcfb42.3.6
[Fix][Connector-V2] Fix connector support SPI but without no args constructor (#6551)https://github.com/apache/seatunnel/commit/5f3c9c36a52.3.5
[Improve][Common] Adapt FILE_OPERATION_FAILED to CommonError (#5928)https://github.com/apache/seatunnel/commit/b3dc0bbc212.3.4
[Improve][Common] Introduce new error define rule (#5793)https://github.com/apache/seatunnel/commit/9d1b2582b22.3.4
[Hotfix][Zeta] Fix conflict dependency of hadoop-hdfs (#4509)https://github.com/apache/seatunnel/commit/66923fbdbd2.3.2
[Improve][build] Give the maven module a human readable name (#4114)https://github.com/apache/seatunnel/commit/d7cd6010512.3.1
[Improve][Project] Code format with spotless plugin. (#4101)https://github.com/apache/seatunnel/commit/a2ab1665612.3.1
[Feature][Connector] add get source method to all source connector (#3846)https://github.com/apache/seatunnel/commit/417178fb842.3.1
[Feature][API & Connector & Doc] add parallelism and column projection interface (#3829)https://github.com/apache/seatunnel/commit/b9164b8ba12.3.1
[Feature][Connector V2] expose configurable options in Hudi (#3383)https://github.com/apache/seatunnel/commit/fd4cec3a952.3.0
fix hudi connector v2 compile error. (#3728)https://github.com/apache/seatunnel/commit/4fba0aa0242.3.0
[Improve][Connector-V2][Hudi] Unified exception for hudi source connector (#3581)https://github.com/apache/seatunnel/commit/b2fda11ddc2.3.0
[bug][Connector-V2][Hudi] HashCode may be negative (#3184)https://github.com/apache/seatunnel/commit/8beffbb6032.3.0
[DEV][Api] Replace SeaTunnelContext with JobContext and remove singleton pattern (#2706)https://github.com/apache/seatunnel/commit/cbf82f755c2.2.0-beta
[#2606]Dependency management split (#2630)https://github.com/apache/seatunnel/commit/fc047be69b2.2.0-beta
[improve][UT] Upgrade junit to 5.+ (#2305)https://github.com/apache/seatunnel/commit/362319ff3e2.2.0-beta
StateT of SeaTunnelSource should extend Serializable (#2214)https://github.com/apache/seatunnel/commit/8c426ef8502.2.0-beta
[Connector-V2] Add Hive sink connector v2 (#2158)https://github.com/apache/seatunnel/commit/23ad4ee7352.2.0-beta
[Connector-V2]Add Hudi Source (#2147)https://github.com/apache/seatunnel/commit/eaedc0a3c72.2.0-beta