Google Cloud Dataflow - Python 将 JSON 流式传输到 PubSub - DirectRunner 和 DataflowRunner 之间的差异答案

【问题标题】：Google Cloud Dataflow - Python Streaming JSON to PubSub - Differences between DirectRunner and DataflowRunnerGoogle Cloud Dataflow - Python 将 JSON 流式传输到 PubSub - DirectRunner 和 DataflowRunner 之间的差异
【发布时间】：2018-06-28 21:35:41
【问题描述】：

试图做一些概念上简单的事情，但我的头撞到了墙上。

我正在尝试在 Python 中创建一个流式数据流作业，它使用来自 PubSub 主题/订阅的 JSON 消息，对每条消息执行一些基本操作（在本例中，将温度从 C 转换为 F），然后发布记录退出另一个话题：

from __future__ import absolute_import

import logging
import argparse
import apache_beam as beam
import apache_beam.transforms.window as window
import json

'''Normalize pubsub string to json object'''
# Lines look like this:
#{"temperature": 29.223036004820123}


def transform_temp(line):
    record = json.loads(line)
    record['temperature'] = record['temperature'] * 1.8 + 32
    return json.dumps(record)

def run(argv=None):
  """Build and run the pipeline."""

  parser = argparse.ArgumentParser()
  parser.add_argument(
      '--output_topic', required=True,
      help=('Output PubSub topic of the form '
            '"projects/<PROJECT>/topic/<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)

  with beam.Pipeline(argv=pipeline_args) as p:
    # Read the pubsub topic into a PCollection.
    lines = ( p | beam.io.ReadStringsFromPubSub(known_args.input_topic)
                | beam.Map(transform_temp)
                | beam.io.WriteStringsToPubSub(known_args.output_topic)
              )

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

使用 DirectRunner 在本地运行此代码时，一切正常。但是，在切换到 DataflowRunner 时，我从未看到任何关于新主题的消息。

我还尝试向 transform_temp 函数添加一些日志记录调用，但在 Dataflow 的控制台日志中看不到任何内容。

有什么建议吗？顺便说一句 - 如果我只是将输入主题与输出主题联系起来，我可以看到消息，所以我知道流工作正常。

非常感谢！

【问题讨论】：

日志中有任何内容吗？此外，您确定您的工作具有访问 Pub/Sub 主题的适当权限吗？当您使用 DirectRunner 在本地运行它时，它可能使用您的个人凭据，而在 Dataflow 上运行时，它使用 GCE 服务帐户。我学到了一个艰难的方法，而不是在缺少权限时引发异常，它只是不发布任何内容。

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

【解决方案1】：

您可能只是缺少一个 windowinto 功能。 Apache Beam 的文档指出，对于流式管道，您需要设置非默认窗口或非默认触发器。由于您尚未定义窗口，因此您有一个全局窗口，因此它可能会在窗口末尾无休止地等待，然后再进入接收器。

【讨论】：