Dataflow处理大数据集高延迟、水印停滞问题排查

基础架构与约束 #

• 管道为混合架构流处理链路：通过Pub/Sub消息传递GCS存储数据文件的路径，设计支持通过Pub/Sub消息动态配置去重规则、分组规则、文件格式（csv、tsv、管道分隔格式等）参数
• 核心业务需求：实现同组数据聚合处理，例如企业销售场景下随消息上报的多类产品数据，需按产品维度归组处理，当前通过side inputs实现对应元素的处理逻辑
• 数据准确性要求：禁止在触发器中配置丢弃迟到数据的规则，全量数据采用Pub/Sub消息的预期watermark

故障现象 #

• 小数据集场景：管道可正常完成处理，仅处理速度偏慢，可正常推进窗口完成计算
• 大数据集场景：管道在GroupByKey步骤挂起，原始GCS文件仅200万行数据时，该步骤会触发多次重复计算，水印长时间无法推进窗口，数据持续积压；提升worker的CPU、内存配置后故障未解决

核心实现代码 #

message = (p 
    | 'Read from Pub/Sub' >> beam.io.ReadFromPubSub(subscription=self._subscription).with_output_types(bytes)
    | 'Convert to JSON' >> beam.Map(lambda message_json: json.loads(message_json))
    | 'Add timestamp to message' >> beam.Map(apply_datetime, key='__message_dt')
    | 'timestamps' >> beam.ParDo(AddTimestampDoFn())
    | 'Window into Fixed Intervals' >> beam.WindowInto(
        windowfn=FixedWindows(600)  # 10 minutes
    ) 

unique_record_keys = message | "Unique Keys" >> beam.Map(
    lambda m: m.get('unique_record_keys'),
)  # 拉取用于首轮去重的属性/键列表

record_grouping_keys = message | "Record Grouping Keys" >> beam.Map(
    lambda m: m.get('record_grouping_keys')
)  # 拉取用于元素分组的键列表，同组元素会以多实体列表形式存储为单个对象

data = message | 'Read all files' >> ReadAllFromDelimitedSource(file_location_key='file_locations')  # 自定义PTransform，本质是指向GCS文件的beam.io.filebasedsource.FileBasedSource，目标是对非规范化文件做规范化处理

key_data = data | "Keyify on Unique" >> beam.ParDo(
    GenerateKey(),
    keys_list=beam.pvalue.AsList(unique_record_keys)
)  # 基于GCS源文件的指定列生成哈希键，用于行级去重

grouped_data = key_data | 'Group Per Unique Key' >> beam.GroupByKey()  # 大数据集场景下该步骤会挂起数小时，水印无法穿过窗口，数据积压不处理

最终排查结果 #

2022-06-05更新：经全链路排查，代码基础运行逻辑无异常；提升内存配置并完成全量数据集校验后，确认故障根因为分组逻辑缺失某一列配置，修正该配置问题后，全量数据可在20分钟内按预期完成处理。

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