如何分析 Python 数据流作业？答案

【问题标题】：How can I profile a Python Dataflow job?如何分析 Python 数据流作业？
【发布时间】：2019-07-08 10:11:16
【问题描述】：

我编写了一个 Python 数据流作业来处理一些数据：

pipeline
| "read" >> beam.io.ReadFromText(known_args.input)  # 9 min 44 sec
| "parse_line" >> beam.Map(parse_line)  # 4 min 55 sec
| "add_key" >> beam.Map(add_key)  # 48 sec
| "group_by_key" >> beam.GroupByKey()  # 11 min 56 sec
| "map_values" >> beam.ParDo(MapValuesFn())  # 11 min 40 sec
| "json_encode" >> beam.Map(json.dumps)  # 26 sec
| "output" >> beam.io.textio.WriteToText(known_args.output)  # 22 sec

（我已删除业务特定语言。）

输入是 1.36 GiB gz 压缩的 CSV，但该作业需要 37 分 34 秒才能运行（我正在使用 Dataflow，因为我预计输入的大小会迅速增长）。

如何识别管道中的瓶颈并加快其执行速度？没有一个单独的函数在计算上是昂贵的。

来自 Dataflow 控制台的自动缩放信息：

12:00:35 PM     Starting a pool of 1 workers. 
12:05:02 PM     Autoscaling: Raised the number of workers to 2 based on the rate of progress in the currently running step(s).
12:10:02 PM     Autoscaling: Reduced the number of workers to 1 based on the rate of progress in the currently running step(s).
12:29:09 PM     Autoscaling: Raised the number of workers to 3 based on the rate of progress in the currently running step(s).
12:35:10 PM     Stopping worker pool.

【问题讨论】：

也许您可以通过 dev@beam.apache.org 询问？

标签： python google-cloud-dataflow

【解决方案1】：

我搜索了dev@beam.apache.org，发现有一个帖子讨论了这个话题：https://lists.apache.org/thread.html/f8488faede96c65906216c6b4bc521385abeddc1578c99b85937d2f2@%3Cdev.beam.apache.org%3E

如果有需要，您可以查看此线程以获取有用信息和/或提出问题/要求/讨论。

【讨论】：

【解决方案2】：

无意中，我发现这种情况下的问题是 CSV 的压缩。

输入是单个 gz 压缩的 CSV。所以我可以更轻松地检查数据，我切换到未压缩的 CSV。这将处理时间减少到 17 分钟以下，并且 Dataflow 的自动缩放在 10 个工作人员时达到峰值。

（如果我仍然需要压缩，我会将 CSV 分成几部分，然后单独压缩。）

【讨论】：

【解决方案3】：

我偶然发现了这个由 Google 提供的 Python Profiler 包：https://cloud.google.com/profiler/docs/profiling-python

【讨论】：