当前位置: 首页 > article >正文

Flink 常用及优化参数

流批模式

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.timeout10 分钟Checkpoint 完成的超时时间,超时则触发失败。SET 'execution.checkpointing.timeout' = '5min';
execution.checkpointing.min-pause0两次 Checkpoint 之间的最小暂停间隔(防止连续触发)。SET 'execution.checkpointing.min-pause' = '30s';
execution.checkpointing.max-concurrent1最大并发 Checkpoint 数量(适用于高吞吐场景)。SET 'execution.checkpointing.max-concurrent' = '2';
execution.checkpointing.externalized-checkpoint-retentionCheckpoint 保留策略:
- DELETE_ON_CANCELLATION(作业取消时删除)
- RETAIN_ON_CANCELLATION(作业取消时保留)
SET 'execution.checkpointing.externalized-checkpoint-retention' = 'RETAIN_ON_CANCELLATION';
state.backend.incrementalfalse是否启用增量 Checkpoint(需使用 RocksDB 状态后端)。SET 'state.backend.incremental' = 'true';
state.checkpoints.dirCheckpoint 存储目录(需配置为持久化路径,如 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

参数调优场景建议

  1. 高吞吐低延迟场景

    • 缩短间隔interval=10s(需权衡 Checkpoint 开销)。
    • 增大并发max-concurrent=2(避免资源争抢)。
    • 增量 Checkpointstate.backend.incremental=true(减少全量快照开销)。
  2. 大状态作业(如窗口聚合)

    • 延长超时timeout=15min(防止大状态序列化超时)。
    • 保留策略externalized-checkpoint-retention=RETAIN_ON_CANCELLATION(便于手动恢复)。
  3. 避免 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 动态调整。


http://www.kler.cn/a/610939.html

相关文章:

  • Serverless架构的应用场景
  • 文件上传的小点总结
  • 自然语言处理(11:RNN(RNN的前置知识和引入)
  • 学习爬虫的第二天——分页爬取并存入表中
  • NO.58十六届蓝桥杯备战|基础算法-枚举|普通枚举|二进制枚举|铺地毯|回文日期|扫雷|子集|费解的开关|Even Parity(C++)
  • Spring MVC 配置详解与入门案例
  • 3ds Max 2026 新功能全面解析
  • husky的简介以及如果想要放飞自我的解决方案
  • Linux centos 7 vsftp本地部署脚本
  • 图灵300题-41~60-笔记003
  • vue3项目使用 python +flask 打包成桌面应用
  • CMake 构建的Qt 项目中的构建套件的配置
  • 《精益创业实战》第2章“对需求性进行压力测试” 总结
  • 破局离散制造:主数据管理驱动数字化转型的实践与启示
  • Git撤回操作全场景指南:未推送与已推送,保留和不保留修改的差异处理
  • 质检LIMS系统在食品生产加工企业的应用 如何保证食品生产企业的安全
  • 区块链技术详解:应用场景与核心优势
  • Python----数据分析(足球运动员数据分析)
  • DeepSeek-V3-0324重磅升级:开源大模型的新标杆
  • 基于PSO粒子群优化的多光谱图像融合算法matlab仿真