CombineFn 中的任务未正确完成 Apache 光束答案

【问题标题】：Task in CombineFn is not properly finalized Apache beamCombineFn 中的任务未正确完成 Apache 光束
【发布时间】：2020-05-13 10:08:45
【问题描述】：

我正在使用 Apache Beam 2.17.0 的 Python 3.7 SDK 进行数据流。代码在本地运行，但我从 pubsub 收集数据。我尝试组合每个键并且一切正常，直到管道调用“merge_accumulators”函数。从此时起，所有底层代码都会执行两次。在调试并深入源码后，我发现任务没有正确完成，这就是为什么它被执行了两次。

这是管道代码：

        options = {
        "runner": "DirectRunner",
        "streaming": True,
        "save_main_session": True
    }

    p = beam.Pipeline(options = PipelineOptions(flags=[], **options))

    processRows = (p
                  |'Read from topic' >> beam.io.ReadFromPubSub(subscription=get_subscription_address())
                  |'Filter do not track' >> beam.ParDo(TakeOutNoTrack)
                  |'Map Data' >> beam.ParDo(mapData)
                   |'Filter metatags' >> beam.ParDo(filterMetatags)
                  |'Label admin' >> beam.ParDo(labelAdmin)
                  |'Process row' >> beam.ParDo(processRow)
                 )

    sessionRow = (processRows
                  |'Add timestamp' >> beam.Map(lambda x: window.TimestampedValue(x, x['timestamp']))
                  |'Key on uuid' >> beam.Map(lambda x: (x['capture_uuid'], x))
                  |'User session window' >> beam.WindowInto(window.Sessions(config_triggers['session_gap']),
                                                              trigger=trigger.AfterWatermark(
                                                                  early=trigger.AfterCount(config_triggers['after_count'])),
                                                              accumulation_mode=trigger.AccumulationMode.ACCUMULATING)
                  |'CombineValues' >> beam.CombinePerKey(JoinSessions())
                 )
    printing = (sessionRow
                |'Printing' >> beam.Map(lambda x: print(x))
                )
    print('running pipeline')
    p.run().wait_until_finish()
    print('done running the pipeline')
    return

这是配置触发器：

    config_triggers = {
    "session_gap": 1320,
    "after_count": 1,
    "session_length": 20
}

这是组合类：

class JoinSessions(beam.CombineFn):
def define_format(self):
    try:
        data = {
            "session_uuid": [],
            "capture_uuid": "",
            "metatags": [],
            "timestamps": [],
            "admin": []
        }

        return data

    except Exception:
        logging.error("error at define data: \n%s" % traceback.format_exc())

def create_accumulator(self):
    try:
        return self.define_format()
    except Exception:
        logging.error("error at create accumulator: \n%s " % traceback.format_exc())

def add_input(self, metatags, input):
    try:
        metatags["session_uuid"].append(input.get('session_uuid'))
        metatags["capture_uuid"] = input.get('capture_uuid')
        metatags["metatags"].append(input.get('metatags'))
        metatags["timestamps"].append(input.get('timestamp'))
        metatags["admin"].append(input.get('admin'))
        print('test add_input')

        return metatags
    except Exception:
        logging.error("error at add input: \n%s" % traceback.format_exc())

def merge_accumulators(self, accumulators):
    # print(accumulators)
    try:
        global test_counter
        tags_accumulated = self.define_format()

        for tags in accumulators:
            tags_accumulated["session_uuid"] += tags['session_uuid']
            tags_accumulated["capture_uuid"] += tags['capture_uuid']
            tags_accumulated["metatags"] += tags['metatags']
            tags_accumulated["timestamps"] += tags['timestamps']
            tags_accumulated["admin"] += tags['admin']
        test_counter += 1
        print('counter = ', test_counter)
        return tags_accumulated

    except Exception:
        logging.error("Error at merge Accumulators: \n%s" % traceback.format_exc())

def extract_output(self, metatags):
    try:
        # print('New input in the pipeline:')
        # print('Extract_output: ')
        # print(metatags, '\n')

        return metatags
    except Exception:
        logging.error("error at return input: \n%s" % traceback.format_exc())

不会抛出错误，也不会抛出异常或某种信息。只是“打印”标签的输出被打印了两次。全局计数器也上升了两次，但管道中只有一个数据条目。 add_input 函数上的打印只执行一次。

我是数据流的新手，所以，如果我犯了一个愚蠢的错误，请见谅。

【问题讨论】：

标签： python google-cloud-dataflow apache-beam

【解决方案1】：

我认为这是由于您设置的触发器。

trigger=trigger.AfterWatermark(early=trigger.AfterCount(config_triggers['after_count'])),accumulation_mode=trigger.AccumulationMode.ACCUMULATING)

config_triggers['after_count] 为 1。

因此，您已经设置了一个触发器，该触发器在每个 1 个元素之后触发，并且您还累积了触发器触发产生的元素。因此，同一窗口中的第二个触发器将包含第一个触发器中的元素，依此类推。有关根据您的用例正确设置管道触发器的详细信息，请参阅以下内容。

https://beam.apache.org/documentation/programming-guide/#triggers

【讨论】：