Flink 配置

Catalog 配置

通过执行以下语句来创建并命名 catalog（请将 <catalog_name> 替换为你的 catalog 名称，将 <config_key>=<config_value> 替换为 catalog 实现的配置项）：

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

以下属性可以全局设置，不限于某种特定的目录（catalog）实现：

如果使用 Hive 目录，可以设置以下属性：

如果使用 Hadoop catalog，可以设置以下属性：

如果使用 REST catalog，可以设置以下属性：

运行时配置

读取选项

Flink 读取选项在配置 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()

对于 Flink SQL，读取选项可以通过 SQL hint 传入，如下所示：

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

选项可以通过 Flink 配置传入，并将应用于当前会话。请注意，并非所有选项都支持这种方式。

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

Read option 优先级最高，其次是 Flink configuration，然后是 Table property。

写入选项

Flink 写入选项在配置 FlinkSink 时传入，如下所示：

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

对于 Flink SQL，可以通过 SQL hint 传入写入选项，如下所示：

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

Range 分布统计类型

配置值为枚举类型：Map、Sketch、Auto。

• Map：为每个键收集精确的采样计数。适用于低基数场景（如几百或几千）。
• Sketch：通过蓄水池采样构建均匀随机采样。非常适合高基数场景（如数百万），因为内存占用保持在较低水平。
• Auto：以 Map 统计开始。但当检测到基数超过阈值（当前为 10,000）时，统计会自动切换为 Sketch。

Range 分布排序键基础权重

range-distribution-sort-key-base-weight：0.0。

如果排序顺序中包含分区列，则每个排序键都会映射到一个分区和一个数据文件。该相对权重可以避免为流量较低的排序键生成过多的小文件。它是一个双精度浮点值，定义每个排序键的最小权重。0.02 表示每个键的基础权重为目标流量权重的 2%（针对每个写入任务）。

例如，目标 Iceberg 表按事件时间进行每日分区。假设数据流包含从现在到 180 天前的事件。按事件时间来看，不同天数之间的流量权重分布通常呈现长尾模式：当天流量最多，越往后的天数（长尾）流量越少。假设写入并行度为 10，全部 180 天的总权重为 10,000，则每个写入任务的目标流量权重为 1,000。假设最久远的 150 天权重之和为 1,000。通常情况下，range 分区器会把最久远的 150 天全部放入同一个写入任务，该任务将写入 150 个小文件（每天一个）。同时保持 150 个打开的文件可能会消耗大量内存，在检查点时刷写并上传 150 个文件（无论多小）也可能非常缓慢。如果将此配置设置为 0.02，则意味着每个排序键的基础权重为目标权重 1,000 的 2%（针对每个写入任务）。这实际上可以避免在同一个写入任务上放置超过 50 个数据文件（每天一个），无论这些文件有多小。

此配置仅适用于低基数场景下的 StatisticsType.Map。对于 StatisticsType.Sketch 的高基数排序列，它们通常不会被用作分区列，否则在写入时会产生过多的分区和小文件。Sketch range 分区器只是将高基数的键拆分为有序的范围。