Dataflow任务在CombinePerKey(max)阶段挂起问题排查求助

这种在Dataflow的CombinePerKey阶段挂起，但DirectRunner运行正常的情况，我之前也碰到过好几次，结合你的场景（从Pub/Sub触发GCS文件读取+窗口聚合求用户最大等级），大概率是以下几个原因导致的，咱们一个个来排查解决：

1. 窗口触发策略未配置，等待水印超时 #

默认情况下，FixedWindow使用的触发是AfterWatermark.pastEndOfWindow()——也就是必须等窗口的**水印（Watermark）**推进到窗口结束时间之后，才会触发聚合计算。但你的数据源是从GCS读取文件，文件内的事件时间可能存在延迟，或者Pub/Sub的水印推进逻辑和本地运行不同，导致水印一直无法到达窗口结束时间，CombinePerKey就会一直处于等待状态，看起来像是挂起。

解决办法：给WindowInto显式配置触发策略，允许基于处理时间提前触发，同时设置迟到数据的容忍时间：

window_events = (
    raw_events | "UseFixedWindow" >> beam.WindowInto(
        beam.window.FixedWindows(5 * 60),
        # 处理时间超过10秒就触发一次聚合
        trigger=beam.trigger.AfterProcessingTime(10),
        # 触发后丢弃旧的累积数据，避免重复计算
        accumulation_mode=beam.trigger.AccumulationMode.DISCARDING,
        # 允许1分钟的迟到数据
        allowed_lateness=beam.window.Duration(1 * 60)
    )
)

2. 事件时间计算错误，导致水印推进异常 #

你的str2timestamp函数使用了time.mktime，这个函数会把解析后的时间转成本地时区的时间戳，但Dataflow内部统一使用UTC时间处理事件时间。如果你的本地时区不是UTC，会导致事件时间被错误偏移，水印始终无法推进到窗口结束时间，聚合任务自然不会执行。

解决办法：修改时间戳转换函数，强制使用UTC时区计算：

def str2timestamp(t, fmt="%Y-%m-%dT%H:%M:%S.%fZ"):
    # 解析UTC时间字符串
    dt = datetime.strptime(t, fmt)
    # 标记为UTC时区后转成时间戳
    return dt.replace(tzinfo=datetime.timezone.utc).timestamp()

3. 数据倾斜或并行度不足 #

如果GCS上的文件过大，或者某个用户的事件量远高于其他用户，会导致Dataflow的某个Worker负载过重，CombinePerKey阶段的某个分组一直无法处理完成，看起来像是整个阶段挂起。而DirectRunner本地运行时资源充足，不会暴露这个问题。

解决办法：

• 拆分大文件：建议将GCS上的文件拆分为100MB以内的小文件，提升读取和处理的并行度；
• 添加Reshuffle：在读取文件后加入Reshuffle步骤，让数据重新均匀分发到各个Worker，避免倾斜：

  raw_event = (
      p | "Read Sub Message" >> beam.io.ReadFromPubSub(topic=args.topic)
        | "Convert Message to JSON" >> beam.Map(lambda message: json.loads(message))
        | "Extract File Name" >> beam.ParDo(ExtractFileNameFn())
        | "Read File from GCS" >> beam.io.ReadAllFromText()
        | "Reshuffle to avoid skew" >> beam.Reshuffle()  # 添加这一步
  )
  

4. 聚合输入的类型问题 #

检查elem['level']的类型，如果是字符串类型，max函数会按照字符串的字典序排序（比如"10"会比"2"小），而且如果存在非数值的level值，可能导致Worker出现隐性错误卡住。虽然DirectRunner可能能处理，但Dataflow的分布式环境下更容易暴露这类问题。

解决办法：在Map阶段将level转换为整数类型：

user_max_level = (
    window_events | 'Group By User ID' >> beam.Map(lambda elem: (elem['user'], int(elem['level'])))
                  | 'Compute Max Level Per User' >> beam.CombinePerKey(max)
)

建议你优先排查事件时间计算和窗口触发策略的问题，这是Dataflow窗口聚合挂起最常见的原因。如果还没解决，再检查数据倾斜和类型问题。

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