Flink Configuration🔗

Catalog Configuration🔗

A catalog is created and named by executing the following query (replace <catalog_name> with your catalog name and <config_key>=<config_value> with catalog implementation config):

CREATE CATALOG <catalog_name> WITH (
  'type'='iceberg',
  `<config_key>`=`<config_value>`
); 

The following properties can be set globally and are not limited to a specific catalog implementation:

The following properties can be set if using the Hive catalog:

The following properties can be set if using the Hadoop catalog:

The following properties can be set if using the REST catalog:

Runtime configuration🔗

Read options🔗

Flink read options are passed when configuring the Flink IcebergSource:

IcebergSource.forRowData()
    .tableLoader(TableLoader.fromCatalog(...))
    .assignerFactory(new SimpleSplitAssignerFactory())
    .streaming(true)
    .streamingStartingStrategy(StreamingStartingStrategy.INCREMENTAL_FROM_SNAPSHOT_ID)
    .startSnapshotId(3821550127947089987L)
    .monitorInterval(Duration.ofMillis(10L)) // or .set("monitor-interval", "10s") \ set(FlinkReadOptions.MONITOR_INTERVAL, "10s")
    .build()

For Flink SQL, read options can be passed in via SQL hints like this:

SELECT * FROM tableName /*+ OPTIONS('monitor-interval'='10s') */
...

Options can be passed in via Flink configuration, which will be applied to current session. Note that not all options support this mode.

env.getConfig()
    .getConfiguration()
    .set(FlinkReadOptions.SPLIT_FILE_OPEN_COST_OPTION, 1000L);
...

Read option has the highest priority, followed by Flink configuration and then Table property.

Write options🔗

Flink write options are passed when configuring the FlinkSink, like this:

FlinkSink.Builder builder = FlinkSink.forRow(dataStream, SimpleDataUtil.FLINK_SCHEMA)
    .table(table)
    .tableLoader(tableLoader)
    .set("write-format", "orc")
    .set(FlinkWriteOptions.OVERWRITE_MODE, "true");

For Flink SQL, write options can be passed in via SQL hints like this:

INSERT INTO tableName /*+ OPTIONS('upsert-enabled'='true') */
...

Range distribution statistics type🔗

Config value is a enum type: Map, Sketch, Auto.

• Map: collects accurate sampling count for every single key. It should be used for low cardinality scenarios (like hundreds or thousands).
• Sketch: constructs a uniform random sampling via reservoir sampling. It fits well for high cardinality scenarios (like millions), as memory footprint is kept low.
• Auto: starts with Maps statistics. But if cardinality is detected higher than a threshold (currently 10,000), statistics are automatically switched to Sketch.

Range distribution sort key base weight🔗

range-distribution-sort-key-base-weight: 0.0.

If sort order contains partition columns, each sort key would map to one partition and data file. This relative weight can avoid placing too many small files for sort keys with low traffic. It is a double value that defines the minimal weight for each sort key. 0.02 means each key has a base weight of 2% of the targeted traffic weight per writer task.

E.g. the sink Iceberg table is partitioned daily by event time. Assume the data stream contains events from now up to 180 days ago. With event time, traffic weight distribution across different days typically has a long tail pattern. Current day contains the most traffic. The older days (long tail) contain less and less traffic. Assume writer parallelism is 10. The total weight across all 180 days is 10,000. Target traffic weight per writer task would be 1,000. Assume the weight sum for the oldest 150 days is 1,000. Normally, the range partitioner would put all the oldest 150 days in one writer task. That writer task would write to 150 small files (one per day). Keeping 150 open files can potentially consume large amount of memory. Flushing and uploading 150 files (however small) at checkpoint time can also be potentially slow. If this config is set to 0.02. It means every sort key has a base weight of 2% of targeted weight of 1,000 for every write task. It would essentially avoid placing more than 50 data files (one per day) on one writer task no matter how small they are.

This is only applicable to StatisticsType.Map for low-cardinality scenario. For StatisticsType.Sketch high-cardinality sort columns, they are usually not used as partition columns. Otherwise, too many partitions and small files may be generated during write. Sketch range partitioner simply splits high-cardinality keys into ordered ranges.