Dataflow使用GroupIntoBatches.WithShardedKey报错求助（Python）

我们有一个Dataflow管道，需根据消息中的动态Key对消息分组后调用外部接口，Key数量未知（已知高频Key但会新增）。原方案通过为不同Key设置固定分片数实现分组，代码及分组函数运行正常。现尝试改用GroupIntoBatches.WithShardedKey，修改后的代码如下：

def make_keys(elem):
    key = elem[0][ATTRIBUTES_FIELD][CODE_ATRIBUTE]
    t = (key, elem)
    return t

def expand(self, pcoll):
    return (pcoll
            | "Add timestamp" >> beam.ParDo(AddTimestamp())
            | "Add Key" >> beam.Map(make_keys)
            | "Shard key" >> beam.GroupIntoBatches.WithShardedKey(self.max_messages, self.max_waiting_time)
            )

但始终出现如下错误：

ValueError: ShardedKeyCoder[TupleCoder[FastPrimitivesCoder, FastPrimitivesCoder]] cannot be made deterministic for 'Window .... messages/Groupby/GroupIntoBatches/ParDo(_GroupIntoBatchesDoFn)'.

请问遗漏了什么配置或步骤？

• 确保分片Key的Coder具备确定性：GroupIntoBatches.WithShardedKey要求分片Key必须使用确定性Coder。当前代码中，make_keys返回的元组包含了整个elem，而elem的嵌套结构默认Coder可能是非确定性的，直接导致ShardedKey的Coder不符合要求。
• 拆分分片Key与业务元素：正确的输入结构应为(shard_key, element)，其中shard_key仅作为分片标识（比如你提取的key字段），不要将整个业务元素合并到分片Key中。调整make_keys仅返回(key, elem)，确保key是基础类型（字符串、整数等自带确定性Coder的类型）。
• 显式注册自定义类型的确定性Coder：如果key是自定义复杂类型，需要通过beam.coders.registry.register_coder为其绑定确定性Coder，比如使用beam.coders.SerializableCoder（需保证类的序列化逻辑完全一致），或改用Protocol Buffers这类自带确定性序列化的格式。
• 检查窗口配置的确定性：确认管道使用的窗口分配器及窗口Coder都是确定性的，避免窗口环节的非确定性配置影响分组过程的Coder校验。

内容的提问来源于stack exchange，提问作者Alex Fragotsis