当前位置：首页 > news >正文

Flink 常用及优化参数

news 来源：原创 2025/5/25 4:54:32

流批模式

SET 'execution.runtime-mode' = 'streaming'; // or batch

基础 Checkpoint 配置

-- 启用 Checkpoint，间隔 5 分钟
SET 'execution.checkpointing.interval' = '5min';
-- Checkpoint 超时时间（10 分钟）
SET 'execution.checkpointing.timeout' = '10min';
-- 最小 Checkpoint 间隔（2.5 分钟）
SET 'execution.checkpointing.min-pause' = '150s';

-- 最大并发 Checkpoint 数（默认 1）
SET 'execution.checkpointing.max-concurrent-checkpoints' = '1';
-- Checkpoint 存储路径（HDFS/S3 等） 配合backend state设置
SET 'state.checkpoints.dir' = 'hdfs:///flink/checkpoints';
-- 作业取消后保留 Checkpoint
SET 'execution.checkpointing.externalized-checkpoint-retention' = 'RETAIN_ON_CANCELLATION';

backend存储路径配置

-- 启用 RocksDB 状态后端
SET 'state.backend' = 'rocksdb'; // hashmap, filesystem

-- RocksDBStateBackend 
-- RocksDB 本地临时存储路径（多磁盘路径用逗号分隔）
SET 'state.backend.rocksdb.localdir' = '/mnt/ssd1/rocksdb,/mnt/ssd2/rocksdb';
-- 增量 Checkpoint（减少存储开销）
SET 'state.backend.rocksdb.incremental' = 'true';

以下是 Flink SQL Client 中与 Checkpoint 配置相关的常用参数及其详细说明，适用于优化容错机制与作业稳定性：

核心 Checkpoint 参数列表

参数名称	默认值	作用描述	配置示例
`execution.checkpointing.interval`	无 (需显式设置)	连续两次 Checkpoint 触发的最小时间间隔（毫秒）。	`SET 'execution.checkpointing.interval' = '60s';`
`execution.checkpointing.timeout`	10 分钟	Checkpoint 完成的超时时间，超时则触发失败。	`SET 'execution.checkpointing.timeout' = '5min';`
`execution.checkpointing.min-pause`	0	两次 Checkpoint 之间的最小暂停间隔（防止连续触发）。	`SET 'execution.checkpointing.min-pause' = '30s';`
`execution.checkpointing.max-concurrent`	1	最大并发 Checkpoint 数量（适用于高吞吐场景）。	`SET 'execution.checkpointing.max-concurrent' = '2';`
`execution.checkpointing.externalized-checkpoint-retention`	无	Checkpoint 保留策略： - `DELETE_ON_CANCELLATION`（作业取消时删除） - `RETAIN_ON_CANCELLATION`（作业取消时保留）	`SET 'execution.checkpointing.externalized-checkpoint-retention' = 'RETAIN_ON_CANCELLATION';`
`state.backend.incremental`	false	是否启用增量 Checkpoint（需使用 RocksDB 状态后端）。	`SET 'state.backend.incremental' = 'true';`
`state.checkpoints.dir`	无	Checkpoint 存储目录（需配置为持久化路径，如 HDFS/S3）。	`SET 'state.checkpoints.dir' = 'hdfs:///flink/checkpoints';`

配置方式

1. 在 SQL Client 中直接设置

-- 启用 Checkpoint 并设置间隔为 1 分钟
SET 'execution.checkpointing.interval' = '60s';
SET 'execution.checkpointing.externalized-checkpoint-retention' = 'RETAIN_ON_CANCELLATION';

2. 通过配置文件 `sql-client-defaults.yaml` 全局配置

execution:
  checkpointing:
    interval: 60s
    timeout: 5min
    min-pause: 30s
    max-concurrent: 2
    externalized-checkpoint-retention: RETAIN_ON_CANCELLATION
state:
  backend: rocksdb
  checkpoints.dir: hdfs:///flink/checkpoints
  backend.incremental: true

参数调优场景建议

高吞吐低延迟场景
- 缩短间隔：interval=10s（需权衡 Checkpoint 开销）。
- 增大并发：max-concurrent=2（避免资源争抢）。
- 增量 Checkpoint：state.backend.incremental=true（减少全量快照开销）。
大状态作业（如窗口聚合）
- 延长超时：timeout=15min（防止大状态序列化超时）。
- 保留策略：externalized-checkpoint-retention=RETAIN_ON_CANCELLATION（便于手动恢复）。
避免 Checkpoint 风暴
- 设置 min-pause：如 min-pause=30s（防止频繁触发导致吞吐下降）。
- 监控背压：Checkpoint 失败可能由数据反压引起，需优化算子并行度。

Checkpoint 状态恢复

从指定 Checkpoint 重启作业（需保留 Checkpoint 元数据）：

# 提交作业时指定恢复路径
./bin/flink run -s hdfs:///flink/checkpoints/<job-id>/chk-1234 ...

注意事项

存储路径权限：确保 Flink 进程对 state.checkpoints.dir 有读写权限（如 HDFS 路径需配置 Kerberos 认证）。
RocksDB 调优：若使用 RocksDB，需额外配置内存参数（如 state.backend.rocksdb.memory.managed=true）。
监控指标：通过 Flink Web UI 监控 Checkpoint 持续时间、大小及失败率，及时调整参数。

通过合理配置 Checkpoint 参数，可显著提升 Flink SQL 作业的容错能力与稳定性，具体数值需根据数据量、集群资源和业务 SLA 动态调整。