Flink SQL

{
  // 开启 Checkpoint
  "execution.checkpointing.enable": "TRUE",
  // Checkpoint 超时时间
  "execution.checkpointing.timeout": "240000",
  // Checkpoint 触发间隔
  "execution.checkpointing.interval": "600000",
  // Checkpoint 存储 namespace，更换路径会清除 state
  "state.checkpoints.namespace": "20230201_01",
  // 使用 RocksDB ssd 做为 state 存储
  "state.backend": "rocksdb",
  // RocksDB memory table 大小
  "state.backend.rocksdb.writebuffer.size": "128mb",
  // RocksDB memory table(mutable and immutable) 总数量, 当写入速度过快，
  // 或者 flush 线程速度较慢，出现 memtable 数量超过了指定大小，请求会无法写入
  "state.backend.rocksdb.writebuffer.count": "2",
  // RocksDB 后台异步刷新线程数，当 mutable memory table 满了之后，会变成
  // immutable memory table，RocksDB 后台会异步的将 immutable memory table 
  // 写到磁盘，生成 SST 文件
  "state.backend.rocksdb.flush.thread.num": "4",
  // RocksDB 后台异步合并线程数，RocksDB 底层实现是 LSTM-Tree，实际生成的 SST 文件是不可修改的，
  // 更新会追加一条新的数据，查询的时候会按照时间排序返回结果，合并线程会定期合并一些 SST 文件的过
  // 期数据
  "state.backend.rocksdb.compaction.thread.num": "4",
  // 是否允许 RocksDB 制作快照期间开启 SST 小文件合并
  "state.backend.state-file-batch.enable": "true",
  // mini-batch 缓存大小，单个 subtask 最大缓存多少数据就会触发一次 operator，如果不开启 mini-batch，
  // 每条数据都会访问 state，增加 state.backend 的访问压力
  "table.exec.mini-batch.size": "5000",
  // mini-batch 最大多少时间触发一次 operator，也是整个 job 数据的最大缓存时间
  "table.exec.mini-batch.allow-latency": "5s",
  "table.exec.hive.infer-source-parallelism.max": "32",
  "taskmanager.memory.framework.heap.size": "128m",
  "taskmanager.memory.jvm-overhead.fraction": "0.3",
  "taskmanager.memory.jvm-overhead.max": "3g",
  "taskmanager.memory.jvm-overhead.min": "2g"
}