Flink 消费 Kafka 配置优化实战分享

Properties props = new Properties();


// 基础连接配置
props.setProperty("bootstrap.servers", "kafka1:9092,kafka2:9092");
props.setProperty("group.id", "flink-group");


// 消费者配置 - 这些都是我调了很久的最佳实践
props.setProperty("fetch.min.bytes", "1048576");   1MB，别设太小，影响吞吐量
props.setProperty("fetch.max.wait.ms", "500");     我之前设1000ms，延迟太高了
props.setProperty("max.partition.fetch.bytes", "5242880");   5MB，根据数据量调整
props.setProperty("max.poll.records", "10000");    单次拉取条数，太大容易OOM
props.setProperty("receive.buffer.bytes", "1048576");   网络接收缓冲区
props.setProperty("send.buffer.bytes", "1048576");    // 网络发送缓冲区


// 消费者重试和超时配置
props.setProperty("session.timeout.ms", "30000");     // 会话超时时间
props.setProperty("heartbeat.interval.ms", "10000");  // 心跳间隔
props.setProperty("max.poll.interval.ms", "300000");  // 最大拉取间隔
props.setProperty("request.timeout.ms", "30000");     // 请求超时时间


// 序列化配置
props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");


// 安全配置（如果需要）
props.setProperty("security.protocol", "SASL_PLAINTEXT");
props.setProperty("sasl.mechanism", "PLAIN");

# JobManager 配置
jobmanager:
  memory:
    process:
      size: 4096m  # 给大点，不然元数据多了扛不住
    jvm-metaspace:
      size: 512m   # 元空间也得够
    jvm-overhead:
      fraction: 0.1


# TaskManager 配置
taskmanager:
  memory:
    process:
      size: 8192m  # 总内存，建议8G起步
    framework:
      heap:
        size: 1024m  # 框架堆内存
    task:
      heap:
        size: 4096m  # 任务堆内存，这个最重要
    managed:
      size: 2048m    # 托管内存，RocksDB用这个
    jvm-metaspace:
      size: 512m     # JVM元空间
    jvm-overhead:
      fraction: 0.1   # JVM开销


  # CPU 配置
  cpu:
    cores: 4.0    # CPU核心数，根据实际机器配置


  # 网络配置
  network:
    memory:
      fraction: 0.1
    request-backoff:
      initial: 100
      max: 10000


# Checkpoint 配置
execution:
  checkpointing:
    interval: 180000           # 3分钟一次，太频繁CPU吃不消
    timeout: 120000           # 超时时间，给够点
    min-pause: 60000         # 最小间隔1分钟
    max-concurrent: 1        # 并发checkpoint数量
    mode: EXACTLY_ONCE      # 精确一次语义
    unaligned: true         # 非对齐checkpoint，建议开启
    externalized:
      enabled: true         # 外部化checkpoint


# State Backend 配置
state:
  backend: rocksdb          # 用RocksDB准没错
  backend.incremental: true # 增量checkpoint
  checkpoints:
    dir: hdfs://namenode:8020/flink/checkpoints  # 存储路径
  savepoints:
    dir: hdfs://namenode:8020/flink/savepoints   # 保存点路径


  # RocksDB 具体配置
  rocksdb:
    block-cache-size: 256mb     # 块缓存大小
    write-buffer-size: 64mb     # 写缓存大小
    max-write-buffer-number: 4  # 最大写缓存数
    block-size: 64kb           # 块大小
    compaction-style: LEVEL    # 压缩方

总内存分配建议：
- 任务堆内存：40-50%
- 托管内存：30-40%
- 框架内存：10-15%
- JVM开销：10-15%

taskmanager:
  memory:
    process:
      size: 8g
    task:
      heap:
        size: 4g
    managed:
      size: 2g

taskmanager:
  memory:
    process:
      size: 16g
    task:
      heap:
        size: 8g
    managed:
      size: 6g

taskmanager:
  memory:
    process:
      size: 32g
    task:
      heap:
        size: 16g
    managed:
      size: 12g

# RocksDB 优化配置
state:
  backend: rocksdb
  backend.memory.managed: true
  rocksdb:
    block-cache-size: 256mb     # 读缓存
    write-buffer-size: 128mb    # 写缓存
    max-write-buffer-number: 4  # 写缓存个数
    
taskmanager:
  memory:
    managed:
      fraction: 0.4            # 托管内存占比提高
    framework:
      heap:
        size: 1g              # 框架内存可以适当减少