使用GCP Dataflow从GCS读取数据的速度非常慢&Apache Beam Python SDK

Question

首先让我先说明一下，我刚刚开始了解如何使用Beam的Python SDK和GCP Dataflow！

问题：我的管道对于几乎所有的用例都非常有效。没有我可以抱怨的错误。我只是对一些可能的瓶颈或我可以做的优化有一些问题。我注意到，当处理大小为50MB的gzipped文件时，我的管道执行时间几乎超过3个小时。不完全确定是否有任何方法来加快这一部分。下面是在作业最终成功完成之前我看到的一堆日志警告的截图。

记录数据流的输出详细信息

以下是管道的相关片段：

if __name__ == '__main__':

    parser = argparse.ArgumentParser(
        description='Beam Pipeline: example'
    )
    parser.add_argument('--input-glob',
                        help='Cloud Path To Raw Data',
                        required=True)
    parser.add_argument('--output-bq-table',
                        help='BQ Native {dataset}.{table}')
    known_args, pipeline_args = parser.parse_known_args()

    with beam.Pipeline(argv=pipeline_args) as p:

        json_raw = (p | 'READING RAW DATA' >> beam.io.ReadFromText(known_args.input_glob)
            | 'JSON-ing' >> beam.Map(lambda e: json.loads(e))
        )

额外信息：

>

我正在使用Airflow的Dataflowhook启动此操作。

我尝试过不同的机器类型，希望在这个问题上投入更多的计算能力可以使它消失，但到目前为止还没有成功。

以下管道执行参数：

--runner=DataflowRunner \
--project=example-project \
--region=us-east4 \
--save_main_session \
--temp_location=gs://example-bucket/temp/ \
--input-glob=gs://example-bucket/raw/  \
--machine_type=n1-standard-16 \
--job_name=this-is-a-test-insert

Answer 1

Beam有许多优化，允许它将文件处理拆分给多个工作人员。此外，它能够拆分单个文件，供多个工作人员并行使用。

不幸的是，这在gzip文件中是不可能的。这是因为gzip文件被压缩到单个块中，必须连续解压缩。当光束工作者读取这个文件时，它必须串行读取整个东西。

有一些压缩格式允许您并行读取它们（这些通常是“多块”格式）。不幸的是，我认为Beam Python SDK目前只支持串行格式。

如果需要以这种方式工作管道，可以尝试在readfromtext之后添加beam.reshuffle操作。通过这样做，您的管道仍将串行读取文件，但并行应用所有下游操作（请参见PTransform guide中的性能部分，以了解为什么会出现这种情况）。

其他一些备选方案：

• 将数据分离到多个gzip文件中。
• 在管道中消耗它之前解压缩它。
• （半开玩笑/半认真）为梁贡献多块压缩支撑？:)):)