添加Reshuffle后BatchElements在DataflowRunner批大小异常如何解决

该问题由DataflowRunner的流水线融合优化机制与Reshuffle的元素分发逻辑冲突导致：

• Reshuffle操作会将上游元素随机打散分发到下游工作实例，针对无界流场景，DataflowRunner默认对Reshuffle下游的节点开启逐元素触发的融合优化，每个元素到达BatchElements节点时会被立即处理，不会等待积累到设置的min_batch_size阈值。而DirectRunner未实现这套生产级的低延迟优化逻辑，所以会严格按照BatchElements的配置积累批次。
• 移除Reshuffle后，DataflowRunner可将Pub/Sub读取、解析、扁平化操作和BatchElements节点融合到同一个工作线程，上游成批拉取的Pub/Sub消息会直接批量进入BatchElements，可轻松达到设定的批次阈值。

• 方案一：使用GroupIntoBatches替代BatchElements
  给每个元素生成范围可控的随机key（比如取值范围0~9的整数），通过GroupIntoBatchesAPI指定批次大小，批次生成完成后丢弃key即可。该API不受融合优化影响，可在Dataflow上严格保证批次大小，示例代码如下：

# Reshuffle 之后替换原有 BatchElements 逻辑
| "Add Random Shard Key" >> Map(lambda elem: (random.randint(0, 9), elem))
| "Generate Fixed Size Batches" >> GroupIntoBatches(batch_size=50)
| "Remove Shard Key" >> FlatMap(lambda key_value: key_value[1])

• 方案二：显式禁止BatchElements节点的流水线融合
  给BatchElements步骤添加资源hint阻止上游融合，强制Dataflow为BatchElements分配独立的处理进程，即可正常积累批次：

| "Batch Typed Element" >> BatchElements(
    min_batch_size=50, max_batch_size=50, target_batch_duration_secs=None
).with_resource_hints(prevent_fusion=True)

• 方案三：调整Reshuffle的执行位置
  如果Reshuffle仅用于打散负载避免数据倾斜，可将Reshuffle操作移到BatchElements之后执行，既可以实现负载均衡，也不会影响批次大小的控制。

内容的提问来源于stack exchange，提问作者Simant Luitel