将数据从 Cloud Storage 读取到 Dataflow

如需将数据从 Cloud Storage 读取到 Dataflow，请使用 Apache Beam TextIO 或 AvroIO I/O 连接器。

添加 Google Cloud 库依赖项

如需将 TextIO 或 AvroIO 连接器与 Cloud Storage 搭配使用，请添加以下依赖项。此库为 "gs://" 文件名提供架构处理程序。

Java

<dependency>
  <groupId>org.apache.beam</groupId>
  <artifactId>beam-sdks-java-io-google-cloud-platform</artifactId>
  <version>${beam.version}</version>
</dependency>

Python

apache-beam[gcp]==VERSION

Go

import _ "github.com/apache/beam/sdks/v2/go/pkg/beam/io/filesystem/gcs"

如需了解详情，请参阅安装 Apache Beam SDK。

最大并行数量

TextIO 和 AvroIO 连接器支持两个并行级别：

  • 各个文件单独键控，以便多个工作器读取。
  • 如果文件未压缩，则连接器可以单独读取每个文件的子范围，从而产生高度并行性。只有当文件中的每一行都是有意义的记录时，才能进行这种拆分。例如，在默认情况下，它不适用于 JSON 文件。

性能

下表展示了从 Cloud Storage 读取数据的性能指标。工作负载使用 Java 版 Apache Beam SDK 2.49.0 在一个 e2-standard2 工作器上运行。它们未使用 Runner v2。

1 亿条记录 | 1 KB | 1 列 吞吐量（字节） 吞吐量（元素）
读取 320 MBps 每秒 320,000 个元素

这些指标基于简单的批处理流水线。它们旨在比较 I/O 连接器之间的性能，不一定代表实际流水线。Dataflow 流水线性能很复杂，它受到多个因素的影响，包括虚拟机类型、正在处理的数据量、外部来源和接收器的性能以及用户代码。指标基于运行 Java SDK，不代表其他语言 SDK 的性能特征。如需了解详情，请参阅 Beam IO 性能。

最佳做法

  • 避免将 watchForNewFiles 与 Cloud Storage 搭配使用。对于大型生产流水线来说，此方法的扩缩性较差，因为连接器必须在内存中保留可见文件的列表。该列表无法从内存中清空，这会随着时间的推移减少工作器的工作记忆。请考虑改用适用于 Cloud Storage 的 Pub/Sub 通知。如需了解详情，请参阅文件处理模式。

  • 如果文件名和文件内容都是有用的数据，请使用 FileIO 类别读取文件名。例如，文件名可能包含在处理文件中的数据时有用的元数据。如需了解详情，请参阅访问文件名。FileIO 文档还介绍了此模式的示例。

后续步骤