如何将Kafka Consumer eachBatch消费批次大小调整为100条

你需要组合调整消费者初始化参数，匹配Kafka拉取的条数、字节数双重阈值规则，才能稳定拿到单批次100条的消息，适配同步BigQuery的高吞吐场景。

核心配置说明 #

• maxPollRecords: 100：最核心的条数控制参数，定义单次poll请求从单个分区最多拉取的消息条数，硬上限设为100
• maxBytesPerPartition：单个分区单次拉取的最大字节阈值，必须设置为大于100条消息的预估总大小，避免没拉够100条就触发字节上限截断。比如单条消息平均大小10KB的话，这个值设为1MB（1024*1024）即可，建议留30%以上冗余应对偶发的大消息
• fetchMaxBytes：单次fetch请求全局最大字节阈值，需大于等于「消费的分区总数 * maxBytesPerPartition」，避免全局字节卡限导致单分区拉不够条数
• fetchWaitMaxMs：broker端等待拉取的最长时间，建议设为300~1000ms，让broker尽量凑够条数再返回，减少过小批次的出现，适配BigQuery批量写入的性能要求

可直接复用的代码示例 #

// 消费者初始化配置
const consumer = kafka.consumer({
  groupId: 'bigquery-sync-group', // 替换成你实际的消费组ID
  maxPollRecords: 100,
  maxBytesPerPartition: 1 * 1024 * 1024, // 按你的实际消息大小调整
  fetchMaxBytes: 10 * 1024 * 1024, // 按实际订阅的分区总数调整
  fetchWaitMaxMs: 500,
  // 其余原有配置保持不变
})

// 消费逻辑
await consumer.run({
  eachBatchAutoResolve: false, // 同步BQ建议关掉自动提交，处理完再手动提交避免丢数
  eachBatch: async ({ batch, resolveOffset, commitOffsetsIfNecessary }) => {
    // 此处batch.messages的长度最多为100条
    // 执行BigQuery批量写入逻辑
    // 写入成功后再提交偏移量
    for (let i = 0; i < batch.messages.length; i++) {
      resolveOffset(batch.messages[i].offset)
    }
    await commitOffsetsIfNecessary()
  }
})

场景适配提示 #

针对每日百万级消息同步BigQuery的场景补充说明：eachBatch的批次是按分区维度拆分的，如果你订阅的topic有N个分区，单次poll最多会拉取N*100条消息，分N次触发eachBatch回调，每次回调对应单个分区的最多100条消息。如果需要攒全局维度的100条再写入BQ，可以在消费逻辑里加一层内存缓冲，攒够条数再触发一次BQ写入，能大幅降低BQ API调用频次，提升整体吞吐。

内容的提问来源于stack exchange，提问作者Rayappan A