從 Cloud Storage 讀取資料至 Dataflow

儲存在 Cloud Storage 中的非結構化或半結構化資料 (例如文字、CSV、JSON 或 Avro 檔案)，通常需要大規模擷取及轉換，才能用於下游分析。將 Cloud Storage 資料擷取至 Dataflow 後，您就能執行分散式平行管道，可靠地讀取及處理批次和串流工作負載中的大量檔案資料集。

如要建構自訂管道，您可以使用 Apache Beam I/O 連接器 (例如 TextIO 或 AvroIO) 讀取檔案進行處理。如果是常見的工作流程，您也可以使用 Google 提供的 Dataflow 範本部署預先建構的管道，不必編寫自訂程式碼。

加入 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 連接器支援兩種平行處理層級：

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

效能

下表列出從 Cloud Storage 讀取的效能指標。這些工作負載是在一個 e2-standard2 worker 上執行，使用 Java 適用的 Apache Beam SDK 2.49.0。他們沒有使用 Portable Runner。

1 億筆記錄 | 1 KB | 1 個資料欄 處理量 (位元組) 處理量 (元素)
閱讀 320 MBps 每秒 320,000 個元素

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

最佳做法

  • 請避免搭配使用 watchForNewFiles 和 Cloud Storage。這種方法不適合大型正式版管道，因為連接器必須在記憶體中保留已查看檔案的清單。清單無法從記憶體中清除，因此 worker 的工作記憶體會隨著時間減少。建議改用 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;
  }
}

後續步驟