数据流//高系统滞后和缓慢的水印

Question

我正在测试数据管道，该数据管道是流媒体管道之间的混合物，并指向GCS数据文件的消息。该系统的目的是允许酒吧子消息决定重复数据删除，组和文件格式的参数（CSV，TSV，管道划界等）。我的系统成功地用于较小的数据集，但似乎正在与较大的数据集斗争。

之所以如此，之所以如此，是因为大多数数据集都可以分组在一起。例如，销售有可能出售消息的公司的公司，我们希望将这些产品分组在一起。我们利用这些元素的一方输入来进行处理。

数据需要准确，因此不允许将丢弃到后期的数据中添加到触发器。当前所有数据都有预期的PubSub消息的水印。

我在实施中是否缺少一些东西，还是我对性能过高的期望？

：

        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'),
        )  # Fetch the list of attributes/key that we can use for first round of deduplication

        record_grouping_keys = message | "Record Grouping Keys" >> beam.Map(
            lambda m: m.get('record_grouping_keys')
        )  # Fetch list of keys for group elements together as they will be stored as an object with multiple entities in a list.

        data = message | 'Read all files' >> ReadAllFromDelimitedSource(file_location_key='file_locations')  # Custom PTransform that essentially is beam.io.filebasedsource.FileBasedSource pointing to a GCS file. This file is a denormalized file and we are trying to normalize it

        key_data = data | "Keyify on Unique" >> beam.ParDo(
            GenerateKey(),
            keys_list=beam.pvalue.AsList(unique_record_keys)
        )  # Generate a key for hashing to remove duplicates rows based on columns of the source files from GCS

        grouped_data = key_data | 'Group Per Unique Key' >> beam.GroupByKey()  # Hangs up here on large dataset for hours on end with watermark not passing through window and data left unprocessed

成功处理的较小数据集：

悬挂的较大数据集（还请注意，它看起来像该步骤一样发射了几次。问题）：

我碰到了内存和CPU，因为它看起来挂在上面，但无济于事。稍后它遇到了同样的问题。

---

更新：2022-06-05

在进一步检查时，该代码正常运行。在增加内存并在数据集上执行完整检查后，在某个列上丢失了组。校正后，数据在20分钟内按预期进行处理。

Answer 1

系统的目的是允许酒吧子消息决定有限数据集的参数及其处理项

我不确定对应用程序的含义，具体取决于系统或如何工作（也许取决于如何您在GenerateKey dofn中消耗侧面输入unique_record_keys）。从管道的外观来看，您正在一次又一次地将带有人工时间戳的任意酒吧/子消息的窗格应用于所有有限的数据。 key_data保持什么？如果您所有的键都是唯一的，那么GBK意味着什么？

明智的表现，您可以尝试

Answer 2

请仔细检查分组键，并确保您不要尝试将仅花费时间完成的数据集分组。

在此问题上遇到的问题是，组键集有错误，并且无法正确分组在唯一元素上，因此所有键都被重复重复，因此整个窗口的数据集被认为是独特的集合。这没有在较小的数据集上显示，仅在消息的数据驱动性质上存在。

在校正组键后，请在钥匙集中仔细检查套管，即上的键盘，该系统能够正确分组到较小的数据集中并正确定义唯一的集合。