Flink Scenario

MOR Table Parameter Simplification and Table Management

create table hudi_mor (Service field)
with (
'connector'='hudi',
'path' = 'hdfs://hacluster/tmp/hudi/stream_mor',
'table.type'='MERGE_ON_READ',
'hoodie.datasource.write.recordkey.field'='Primary key field',
'hoodie.bucket.index.num.buckets'='20',
'hoodie.managed.by.ldms'='true',
'hive_sync.enable' = 'true',
'hive_sync.table' = 'Name of the table to be synchronized to Hive',
'hive_sync.db' = 'Name of the database to be synchronized to Hive',
'hive_sync.mode' = 'hms',
'hive_sync.metastore.uris' = 'Value of hive.metastore.uris in the hive-site.xml file on the Hive client',
'properties.hive.metastore.kerberos.principal' = 'Value of hive.metastore.kerberos.principal in the hive-site.xml file on the Hive client'
)

When Flink writes data to MOR tables, if the MRS cluster includes the LDMS service, add hoodie.managed.by.ldms to the WITH attribute of the Sink Hudi table (Hive synchronization must also be configured). This setting automatically configures the MOR tables to generate only a compaction plan and disables clean and archive operations. The LDMS service then manages the compaction, clean, and archive operations for the tables. In addition, the following parameters are automatically added to set the write.precombine.field attribute to the primary key field (using the first field for composite keys), set the index to BUCKET, set the write mode to LSM, set the Parquet file compression format to zstd, and enable the log index. These settings improve read/write performance and optimize storage. The preceding table creation statements achieve an equivalent effect as follows:

create table hudi_mor (Service field)
with (
'connector'='hudi',
'path' = 'hdfs://hacluster/tmp/hudi/stream_mor',
'table.type'='MERGE_ON_READ',
'hoodie.datasource.write.recordkey.field'='Primary key field',
'write.precombine.field'='Primary key field',
'index.type'='BUCKET',
'hoodie.bucket.index.num.buckets'='20',
'hoodie.managed.by.ldms'='true',
'hoodie.lsm.style'='true',
'hoodie.parquet.compression.codec'='zstd',
'hoodie.log.index.enable'='true',
'hoodie.datasource.write.hive_style_partitioning'='true',
'compaction.delta_commits'='10',
'compaction.async.enabled'='false',
'compaction.schedule.enabled'='true',
'clean.async.enabled'='false',
'hoodie.clean.automatic'='false',
'hoodie.archive.automatic'='false',
'hive_sync.enable' = 'true',
'hive_sync.table' = 'Name of the table to be synchronized to Hive',
'hive_sync.db' = 'Name of the database to be synchronized to Hive',
'hive_sync.mode' = 'hms',
'hive_sync.metastore.uris' = 'Value of hive.metastore.uris in the hive-site.xml file on the Hive client',
'properties.hive.metastore.kerberos.principal' = 'Value of hive.metastore.kerberos.principal in the hive-site.xml file on the Hive client'
)

COW Table Parameter Simplification and Table Management

create table hudi_cow (Service field,PRIMARY KEY (Primary key field) NOT ENFORCED)
with (
'connector'='hudi',
'path' = 'hdfs://hacluster/tmp/hudi/stream_cow',
'hoodie.managed.by.ldms'='true',
'hive_sync.enable' = 'true',
'hive_sync.table' = 'Name of the table to be synchronized to Hive',
'hive_sync.db' = 'Name of the database to be synchronized to Hive',
'hive_sync.mode' = 'hms',
'hive_sync.metastore.uris' = 'Value of hive.metastore.uris in the hive-site.xml file on the Hive client',
'properties.hive.metastore.kerberos.principal' = 'Value of hive.metastore.kerberos.principal in the hive-site.xml file on the Hive client'
)

When Flink writes data to COW tables, if the MRS cluster includes the LDMS service, add hoodie.managed.by.ldms to the WITH attribute of the Sink Hudi table (Hive synchronization must also be configured). This setting automatically disables clean and archive operations for COW tables. The LDMS service then manages the clean and archive operations for the tables. In addition, the following parameters are automatically added to set write.precombine.field as the primary key field (using the first composite keys), set the index to MEMORY, and set the Parquet file compression format to zstd. The preceding table creation statements achieve an equivalent effect as follows:

create table hudi_cow (Service field,PRIMARY KEY (Primary key field) NOT ENFORCED)
with (
'connector'='hudi',
'path' = 'hdfs://hacluster/tmp/hudi/stream_cow',
'write.operation' = 'upert', --The update mode is used by default. You can set the mode to insert to enable the append mode.
'index.type' = 'MEMORY',
'write.precombine.field'='Primary key field',
'hoodie.parquet.compression.codec'='zstd',
'hoodie.managed.by.ldms'='true',
'clean.async.enabled'='false',
'hoodie.clean.automatic'='false',
'hoodie.archive.automatic'='false',
'hive_sync.enable' = 'true',
'hive_sync.table' = 'Name of the table to be synchronized to Hive',
'hive_sync.db' = 'Name of the database to be synchronized to Hive',
'hive_sync.mode' = 'hms',
'hive_sync.metastore.uris' = 'Value of hive.metastore.uris in the hive-site.xml file on the Hive client',
'properties.hive.metastore.kerberos.principal' = 'Value of hive.metastore.kerberos.principal in the hive-site.xml file on the Hive client'
)

Table Parameter Simplification and Table Management in Append Mode

In Append mode, COW tables are used for writing, and the high-performance small file merging tool (see section 15.2.4.17) and LDMS are used to automatically manage the clean/archive operation.

create table hudi_cow (Service field,PRIMARY KEY (Primary key field) NOT ENFORCED)
with (
'connector'='hudi',
'path' = 'hdfs://hacluster/tmp/hudi/stream_cow',
'write.operation' = 'insert',
'hoodie.managed.by.ldms'='true',
'hive_sync.enable' = 'true',
'hive_sync.table' = 'Name of the table to be synchronized to Hive',
'hive_sync.db' = 'Name of the database to be synchronized to Hive',
'hive_sync.mode' = 'hms',
'hive_sync.metastore.uris' = 'Value of hive.metastore.uris in the hive-site.xml file on the Hive client',
'properties.hive.metastore.kerberos.principal' = 'Value of hive.metastore.kerberos.principal in the hive-site.xml file on the Hive client'
)