# Paimon配置与类型 #### WITH参数说明 本章节介绍Flink连接器读写Paimon表的参数配置。通过配置相关参数，可以控制Flink与Paimon表格式之间的连接和读写行为，满足不同业务场景下的数据读写需求。 - **Source参数** 表1**Source参数** | 参数 | 是否必填 | 类型 | 默认值 | 取值范围 | 说明 | |:---|:---|:---|:---|:---|:---| | scan.mode | 否 | String | default | default / latest-full / latest / compacted-full / from-snapshot / from-snapshot-full / from-timestamp / from-creation-timestamp / incremental | 指定读取模式。参数取值如下： - default自动判断。 - latest-full先产出最新快照再持续读取变更。 - latest仅持续读取最新变更（不读历史）。 - compacted-full产出最近压缩后的快照。 - from-snapshot从指定快照开始（不产出初始快照）。 - from-snapshot-full从指定快照开始（含初始快照）。 - from-timestamp从指定时间戳开始。 - from-creation-timestamp从指定创建时间戳开始。 - incremental增量读取两个快照间变更。 Flink运行时根据此模式决定Source的数据扫描策略和起始位置，影响首次读取的数据范围。 | | scan.snapshot-id | 否 | Long | 无 | 正整数 | 指定起始快照ID。仅在scan.mode为from-snapshot时生效。Flink运行时从该快照对应的表状态开始读取数据，快照ID不存在时作业将启动失败。 | | scan.timestamp-millis | 否 | Long | 无 | 毫秒时间戳 | 指定起始时间戳。仅在scan.mode为from-timestamp时生效。Flink运行时从该时间戳对应的快照开始读取数据，自动选择不晚于该时间戳的最新快照作为起始点。 | | scan.bounded.watermark-interval | 否 | Duration | 无 | 正整数 | 设置有界读取的水印间隔。用于有界流模式的Watermark推进控制。Flink运行时按此间隔推进Watermark，当Watermark超过数据范围时Source自动结束。 | | lookup.cache.max-rows | 否 | Integer | 无 | 正整数 | 设置维表查询缓存的最大行数。仅对Lookup JOIN场景生效。Flink运行时在内存中维护LRU缓存，缓存行数达到此上限后淘汰最久未访问的记录，增大此值可提高查询命中率但增加内存消耗。 | | lookup.cache.ttl | 否 | Duration | 无 | 正整数+时间单位 | 设置维表查询缓存的TTL（存活时间）。仅对Lookup JOIN场景生效。Flink运行时在缓存记录超过TTL后将其失效，下次访问时重新从Paimon存储加载，用于平衡数据时效性与查询性能。 | - **Sink参数** 表2**Sink参数** | 参数 | 是否必填 | 类型 | 默认值 | 取值范围 | 说明 | |:---|:---|:---|:---|:---|:---| | write-mode | 否 | String | auto | auto / append-only / change-log | 指定写入模式。参数取值如下： - auto根据是否有主键自动选择。 - append-only追加写入（无需主键）。 - change-log变更日志写入（需主键表）。 Flink运行时根据此模式决定写入器的类型和数据的处理方式，auto模式下有主键自动使用change-log，无主键使用append-only。 | | merge-engine | 否 | String | deduplicate | deduplicate / partial-update / aggregation / first-row | 指定合并引擎类型，仅主键表生效。 - deduplicate仅保留最新记录。 - partial-update部分字段更新（NULL不覆盖非NULL）。 - aggregation按聚合函数更新。 - first-row去重保留首行（不支持DELETE和UPDATE_BEFORE记录）。 Flink运行时在合并同主键记录时根据此引擎决定合并策略，影响数据的最终写入结果。partial-update和aggregation流式查询时需配合lookup或full-compaction的changelog-producer。 | | write-buffer-size | 否 | MemorySize | 256mb | 正整数 | 设置写入缓冲区大小。 Flink运行时在内存中缓冲写入数据，缓冲区满后刷写到磁盘，增大此值可减少刷写频率提高写入吞吐但增加内存占用。 | | write.max-writers-per-task | 否 | Integer | 1 | 正整数 | 设置每个任务的最大写入器数量。 Flink运行时在每个Sink子任务中创建此数量的写入器并行写入数据，增大此值可提高单任务写入并行度但增加内存和文件句柄消耗。 | | changelog-producer | 否 | String | none | none / input / lookup / full-compaction | 指定变更日志生产者模式。 启用后可产生完整的变更日志供下游流式消费。none不产生；input直接传递输入记录；lookup通过查找产生；full-compaction在全量压缩时产生。Flink运行时在提交数据时根据此模式生成对应的变更日志，供下游Source增量消费。 | | sink.parallelism | 否 | Integer | 无 | 正整数 | 指定Sink算子的并行度。 未设置时使用作业全局并行度。Flink运行时根据此值创建对应数量的Sink子任务，每个子任务负责写入一部分分桶的数据，建议与bucket数保持一致以避免数据倾斜。 | | compaction.optimization-enabled | 否 | Boolean | false | true / false | 控制是否启用压缩优化。 启用后Flink运行时在写入过程中自动触发小文件合并，减少文件数量提高查询性能，但会增加写入时的计算开销。 | | snapshot.time-retained | 否 | Duration | 1h | 正整数 | 设置快照保留时长。 Flink运行时在快照超过此时长后自动清理，释放存储空间。增大此值可保留更多历史数据用于增量读取和时间旅行查询，但会增加存储消耗。 | | snapshot.num-retained.min | 否 | Integer | 10 | 正整数 | 设置最少保留的快照数量。即使快照超过保留时长，也至少保留此数量的快照。Flink运行时在清理快照时以此值为下限，确保增量读取有足够的快照可用。 | | snapshot.num-retained.max | 否 | Integer | 无 | 正整数 | 设置最多保留的快照数量。设置后Flink运行时在快照数量超过此上限时清理最旧的快照，防止快照无限增长占用存储空间。 | - **序列字段参数** 表3**序列字段参数** | 参数 | 是否必填 | 类型 | 默认值 | 取值范围 | 说明 | |:---|:---|:---|:---|:---|:---| | sequence.field | 否 | String | 无 | 有效的列名 | 指定主键表的排序字段。 Flink运行时根据此字段的值判断记录的新旧顺序，值越大表示记录越新。未指定时按数据到达顺序决定。适用于多流更新场景中解决记录乱序问题。 | | rowkind.field | 否 | String | 无 | 有效的列名 | 指定行类型字段，字段值为 '+I'（插入）、'-U'（更新前）、'+U'（更新后）、'-D'（删除）。Flink运行时根据此字段值确定记录的变更类型，无需依赖Flink运行时的回撤机制，适用于从Kafka等外部系统直接读取变更记录的场景。 | - **Lookup参数** 表4**Lookup参数** | 参数 | 是否必填 | 类型 | 默认值 | 取值范围 | 说明 | |:---|:---|:---|:---|:---|:---| | lookup.async | 否 | Boolean | false | true / false | 是否启用异步Lookup。 启用后Flink运行时使用异步线程池并发查询Paimon存储，可显著提高Lookup JOIN的吞吐量，但会增加内存消耗。 | | lookup.async-thread-number | 否 | Integer | 16 | 正整数 | 异步Lookup的线程池大小。 仅在lookup.async为true时生效。Flink运行时使用此数量的线程并发执行Lookup查询，增大此值可提高并发查询能力但增加内存和CPU消耗。 | | lookup.cache | 否 | String | AUTO | AUTO / FULL / MEMORY | Lookup缓存模式。 - AUTO自动选择； - FULL全量缓存（小表推荐）； - MEMORY仅内存缓存。 Flink运行时根据此模式决定缓存策略，FULL模式在首次加载时读取全表数据到缓存，后续查询直接命中缓存。 | - **分桶参数** 表5**分桶参数** | 参数 | 是否必填 | 类型 | 默认值 | 取值范围 | 说明 | |:---|:---|:---|:---|:---|:---| | bucket | 否 | Integer | -1 | -2 / -1 / 正整数 | 设置分桶数量。 - -1为Dynamic Bucket（默认，自动管理分桶）； - -2为Postpone Bucket（延迟分桶，仅Append表）； - 正整数为固定分桶数。 主键表使用固定分桶时建议设置为与Sink并行度相同的值以避免数据倾斜。Flink运行时根据分桶键的哈希值将数据分配到对应桶中，分桶数影响数据分布的均衡性和并行读取能力。 | | bucket-key | 否 | String | 主键列 | 有效的列名 | 指定分桶键。未指定时使用主键列作为分桶键。仅主键表生效。Flink运行时根据此列的哈希值进行分桶路由，选择与业务数据分布均匀的列作为分桶键可提高写入和查询性能。 | #### 数据类型 | Flink Data Type | **Paimon Data Type** | |:---|:---| | BOOLEAN | BOOLEAN | | TINYINT | TINYINT | | SMALLINT | SMALLINT | | INT | INT | | BIGINT | BIGINT | | FLOAT | FLOAT | | DOUBLE | DOUBLE | | DECIMAL | DECIMAL | | STRING | STRING | | CHAR | CHAR | | DATE | DATE | | TIME | TIME | | TIMESTAMP | TIMESTAMP | | TIMESTAMP WITH LOCAL TIMEZONE | TIMESTAMP WITH LOCAL TIMEZONE | | BINARY | BINARY | | BYTES | BYTES |