我有一个简单的管道，从PubSub接收数据，打印它，然后每隔10秒向GroupByKey发射一个窗口，并再次打印该消息。

然而，这个窗口有时似乎会延迟。这是谷歌的限制还是我的代码有问题：

 with beam.Pipeline(options=pipeline_options) as pipe:
        messages = (
            pipe
            | beam.io.ReadFromPubSub(subscription=known_args.input_subscription).with_output_types(bytes)
            | 'decode' >> beam.Map(lambda x: x.decode('utf-8'))
            | 'Ex' >> beam.ParDo(ExtractorAndPrinter())
            | beam.WindowInto(window.FixedWindows(10), allowed_lateness=0, accumulation_mode=AccumulationMode.DISCARDING, trigger=AfterProcessingTime(10) )
            | 'group' >> beam.GroupByKey()
            | 'PRINTER' >> beam.ParDo(PrinterWorker()))

编辑最新代码。我删除了触发器，但问题仍然存在：

class ExtractorAndCounter(beam.DoFn):
    def __init__(self):
        beam.DoFn.__init__(self)

    def process(self, element, *args, **kwargs):
        import logging
        logging.info(element)
        return [("Message", json.loads(element)["Message"])]

class PrinterWorker(beam.DoFn):
    def __init__(self):
        beam.DoFn.__init__(self)

    def process(self, element, *args, **kwargs):
        import logging
        logging.info(element)
        return [str(element)]


class DefineTimestamp(beam.DoFn):
    def process(self, element, *args, **kwargs):
        from datetime import datetime
        return [(str(datetime.now()), element)]

def run(argv=None, save_main_session=True):
    """Build and run the pipeline."""
    parser = argparse.ArgumentParser()
    parser.add_argument(
      '--output_topic',
      required=True,
      help=(
          'Output PubSub topic of the form '
          '"projects/<PROJECT>/topics/<TOPIC>".'))
    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument(
      '--input_topic',
      help=(
          'Input PubSub topic of the form '
          '"projects/<PROJECT>/topics/<TOPIC>".'))
    group.add_argument(
      '--input_subscription',
      help=(
          'Input PubSub subscription of the form '
          '"projects/<PROJECT>/subscriptions/<SUBSCRIPTION>."'))
    known_args, pipeline_args = parser.parse_known_args(argv)

pipeline_options = PipelineOptions(pipeline_args)
pipeline_options.view_as(SetupOptions).save_main_session = save_main_session
pipeline_options.view_as(StandardOptions).streaming = True
with beam.Pipeline(options=pipeline_options) as pipe:
    messages = (
        pipe
        | beam.io.ReadFromPubSub(subscription=known_args.input_subscription).with_output_types(bytes)
        | 'decode' >> beam.Map(lambda x: x.decode('utf-8'))
        | 'Ex' >> beam.ParDo(ExtractorAndCounter())
        | beam.WindowInto(window.FixedWindows(10))
        | 'group' >> beam.GroupByKey()
        | 'PRINTER' >> beam.ParDo(PrinterWorker())
        | 'encode' >> beam.Map(lambda x: x.encode('utf-8'))
        | beam.io.WriteToPubSub(known_args.output_topic))



if __name__ == '__main__':
    logging.getLogger().setLevel(logging.INFO)
    run()