從 Cloud Storage 讀取資料至 Dataflow

如要將資料從 Cloud Storage 讀取至 Dataflow，請使用 Apache Beam TextIO 或 AvroIO I/O 連接器。

加入 Google Cloud 程式庫依附元件

如要搭配 Cloud Storage 使用 TextIO 或 AvroIO 連接器，請加入下列依附元件。這個程式庫提供 "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」。

在 Dataflow 上啟用 Apache Beam I/O 連接器的 gRPC

您可以透過 Dataflow 上的 Apache Beam I/O 連接器，使用 gRPC 連線至 Cloud Storage。gRPC 是 Google 開發的高效能開放原始碼遠端程序呼叫 (RPC) 架構，可用於與 Cloud Storage 互動。

如要加快 Dataflow 工作對 Cloud Storage 的讀取要求，您可以在 Dataflow 上啟用 Apache Beam I/O 連接器，使用 gRPC。

指令列

  1. 請確認您使用的是 Apache Beam SDK 2.55.0 以上版本。
  2. 如要執行 Dataflow 工作，請使用 --additional-experiments=use_grpc_for_gcs 管道選項。如要瞭解不同的管道選項，請參閱「選用標記」。

Apache Beam SDK

  1. 請確認您使用的是 Apache Beam SDK 2.55.0 以上版本。
  2. 如要執行 Dataflow 工作，請使用 --experiments=use_grpc_for_gcs 管道選項。如要瞭解不同的管道選項，請參閱基本選項。

您可以在 Dataflow 上設定 Apache Beam I/O 連接器，在 Cloud Monitoring 中產生 gRPC 相關指標。gRPC 相關指標可協助您執行下列操作：

  • 監控及最佳化傳送至 Cloud Storage 的 gRPC 要求效能。
  • 排解及偵錯問題。
  • 深入瞭解應用程式的使用情形和行為。

如要瞭解如何在 Dataflow 上設定 Apache Beam I/O 連接器，以產生 gRPC 相關指標，請參閱「使用用戶端指標」。 如果您的用途不需要收集指標，可以選擇停用指標收集功能。 如需操作說明，請參閱「停用用戶端指標」。

平行處理工作數量

TextIO 和 AvroIO 連接器支援兩個層級的平行處理：

  • 個別檔案會分別加上索引鍵，因此多個工作站都能讀取。
  • 如果檔案未經壓縮，連接器可以分別讀取每個檔案的子範圍，進而達到非常高的平行處理層級。只有在檔案中的每一行都是有意義的記錄時，才能進行這項分割作業。舉例來說，JSON 檔案預設不會提供這項功能。

成效

下表顯示從 Cloud Storage 讀取資料的效能指標。這些工作負載是在一個 e2-standard2 工作站上執行，並使用 Java 適用的 Apache Beam SDK 2.49.0。他們沒有使用 Runner v2。

1 億筆記錄 | 1 kB | 1 個資料欄 處理量 (位元組) 處理量 (元素)
已讀 320 MBps 每秒 32 萬個元素

這些指標是以簡單的批次管道為依據。這些基準旨在比較 I/O 連接器之間的效能，不一定代表實際的管道。Dataflow 管道效能相當複雜，取決於 VM 類型、處理的資料、外部來源和接收器的效能，以及使用者程式碼。這些指標是根據執行 Java SDK 取得，無法代表其他語言 SDK 的效能特徵。詳情請參閱「Beam IO 效能」。

最佳做法

  • 避免將 watchForNewFiles 與 Cloud Storage 搭配使用。這種做法不適合大型製作管線，因為連接器必須將已查看的檔案清單保留在記憶體中。這份清單無法從記憶體中清除，因此工作站的工作記憶體會隨著時間減少。建議改用 Cloud Storage 的 Pub/Sub 通知。詳情請參閱「檔案處理模式」。

  • 如果檔案名稱和檔案內容都是實用資料，請使用 FileIO 類別讀取檔案名稱。舉例來說，檔案名稱可能包含處理檔案中資料時有用的中繼資料。詳情請參閱「存取檔案名稱」。FileIO 說明文件也顯示了這個模式的範例。

範例

下列範例說明如何從 Cloud Storage 讀取資料。

Java

如要向 Dataflow 進行驗證，請設定應用程式預設憑證。 詳情請參閱「為本機開發環境設定驗證」。

import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.values.TypeDescriptors;

public class ReadFromStorage {
  public static Pipeline createPipeline(Options options) {
    var pipeline = Pipeline.create(options);
    pipeline
        // Read from a text file.
        .apply(TextIO.read().from(
            "gs://" + options.getBucket() + "/*.txt"))
        .apply(
            MapElements.into(TypeDescriptors.strings())
                .via(
                    (x -> {
                      System.out.println(x);
                      return x;
                    })));
    return pipeline;
  }
}

後續步驟