聚集步骤在本地发出结果，但在GCP数据流中部署时不发出结果

Question

(
    p
    | 'read_telemetry_from_pubsub' >> ReadFromPubSub(topic=PUBSUB_TOPIC)
    | 'window_telemetry' >> beam.WindowInto(beam.window.FixedWindows(WINDOW_SIZE))
    | 'format_telemetry_for_influx' >> beam.ParDo(FormatInfluxDB())
    | "add_dummy_key" >> beam.Map(lambda elem: (None, elem))
    | "groupby_dummy_key" >> beam.GroupByKey()
    | "delete_dummy_key" >> beam.MapTuple(lambda _, val: val)
    | "aggregate" >> beam.ParDo(Aggregator())
    | "write_processed_messages_to_influx" >> beam.ParDo(WriteToInfluxDB())
)

这是一个在15分钟的窗口中聚合包含所有消息的列表的类：

class Aggregator(beam.DoFn):
    def process(self, elements):
        # parsing the message list into a pandas DataFrame
        # some preprocessing and agregation steps
        # returns a list with json messages
        return [aggregated_values]

我们使用GCP pub/sub模拟器在本地测试这段代码，它工作得很好。然而，当我们部署到GCP Dataflow时，它不会发出任何结果，也不会在日志中发现错误。此外，我们看到数据新鲜度无限增长。

我们认为我们缺少一些触发函数，但我们不确定这是否是进行此类聚合的正确方法，因为它在本地发出结果，但在部署时不发出结果。当我们使用与默认触发器不同的触发器时，本地我们没有发射。

Answer 1

对于所需的包，使用setup.py文件而不是requirements.txt文件解决了这个问题。此外，我们将python Apache Beam SDK版本从2.29.0更新到2.32.0，现在可以正确地聚合和发出结果。

我们在日志GCP模块中发现了这个错误：

同步pod...（“pipeline_name”）时出错，跳过:[用crashloopbackoff为“sdk0”“startcontainer”失败：“back-off 5m0s restarting failed container=sdk0 pod=pipeline_name)”