Google 會運用 AI 技術將內容翻譯成你偏好的語言，但可能會出錯。

從 Dataflow 寫入 BigQuery

本文說明如何將資料從 Dataflow 寫入 BigQuery。

總覽

在大多數情況下，建議使用代管 I/O 寫入 BigQuery。代管 I/O 提供自動升級和一致的設定 API 等功能。寫入 BigQuery 時，Managed I/O 會自動為批次或串流工作選擇最佳寫入方法。

如需更進階的效能調整，建議使用 BigQueryIO 連接器。詳情請參閱本文的「使用 BigQueryIO 連接器」一節。

效能

下表列出各種工作負載的成效指標。這些工作負載是在一個 e2-standard2 worker 上執行，使用 Java 適用的 Apache Beam SDK 2.49.0。他們沒有使用 Runner v2。

1 億筆記錄 \| 1 KB \| 1 個資料欄	處理量 (位元組)	處理量 (元素)
Storage Write	55 MB/秒	每秒 54,000 個元素
Avro 載入	78 MB/秒	每秒 77,000 個元素
Json Load	54 MB/秒	每秒 53,000 個元素

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

使用 `BigQueryIO` 連接器

BigQuery I/O 連接器支援下列寫入 BigQuery 的方法：

STORAGE_WRITE_API。在這個模式中，連接器會使用 BigQuery Storage Write API (gRPC)，直接寫入 BigQuery 儲存空間。Storage Write API (gRPC) 將串流擷取和批次載入功能合併為單一的高效能 API。這個模式可保證語意只會執行一次。
STORAGE_API_AT_LEAST_ONCE。這個模式也使用 Storage Write API (gRPC)，但提供至少一次語意。這個模式可降低大多數管道的延遲時間。不過，寫入作業可能會重複。
FILE_LOADS。在這個模式中，連接器會將輸入資料寫入 Cloud Storage 的暫存檔案。接著，系統會執行 BigQuery 載入工作，將資料載入 BigQuery。這個模式是有界限 PCollections 的預設模式，最常見於批次管道。
STREAMING_INSERTS。在這個模式中，連接器會使用舊版串流 API。這個模式會盡可能簡化，因此可能會產生一些重複項目。這是無限制 PCollections 的預設模式，但不建議用於新專案。

選擇寫入方法時，請考量下列幾點：

如果是串流工作，建議使用 STORAGE_WRITE_API 或 STORAGE_API_AT_LEAST_ONCE，因為這些模式會直接寫入 BigQuery 儲存空間，不使用中繼暫存檔案。
如果使用至少一次串流模式執行管道，請將寫入模式設為 STORAGE_API_AT_LEAST_ONCE。這項設定更有效率，且符合至少一次串流模式的語意。FILE_LOADS 寫入方法與「至少一次」串流模式不相容。
檔案載入和 Storage Write API (gRPC) 的配額和限制不同。
載入工作會使用共用的 BigQuery 運算單元集區或預留運算單元。如要使用預留運算單元，請在預留項目指派類型為 PIPELINE 的專案中執行載入工作。如果使用共用 BigQuery 運算單元集區，載入工作不需付費。不過，BigQuery 不保證共用集區的可用容量。詳情請參閱「預留項目簡介」。

平行處理工作數量

對於串流管道中的 FILE_LOADS 和 STORAGE_WRITE_API，連接器會將資料分割成多個檔案或串流。一般來說，建議呼叫 withAutoSharding 啟用自動資料分割。
如果是批次管道中的 FILE_LOADS，連接器會將資料寫入分區檔案，然後平行載入 BigQuery。
在批次管道中，STORAGE_WRITE_API 的每個 worker 會根據 shard 總數，建立一或多個串流，寫入 BigQuery。
STORAGE_API_AT_LEAST_ONCE 只有一個預設寫入串流。多個 worker 會附加至這個串流。

最佳做法

Storage Write API (gRPC) 有配額限制。連接器會處理大多數管道的這些限制。不過，在某些情況下，可用的 Storage Write API (gRPC) 串流可能會用盡。舉例來說，如果管道使用自動資料分割和自動調度資源功能，且目的地數量龐大，就可能發生這個問題，尤其是在工作負載變化很大的長時間執行工作中。如果發生這個問題，建議使用 STORAGE_WRITE_API_AT_LEAST_ONCE，即可避免這個問題。
使用 Google Cloud 指標監控 Storage Write API (gRPC) 配額用量。
使用檔案載入時，Avro 的效能通常優於 JSON。如要使用 Avro，請呼叫 withAvroFormatFunction。
根據預設，載入工作會在與 Dataflow 工作相同的專案中執行。如要指定其他專案，請呼叫 withLoadJobProjectId。
使用 Java SDK 時，建議建立代表 BigQuery 資料表結構定義的類別。接著，在管道中呼叫 useBeamSchema，即可自動在 Apache Beam Row 和 BigQuery TableRow 類型之間轉換。如需結構定義類別範例，請參閱 ExampleModel.java。
如果載入的資料表含有複雜的結構定義，且包含數千個欄位，建議呼叫 withMaxBytesPerPartition，為每項載入工作設定較小的大小上限。
根據預設，BigQueryIO 會使用適合大多數管道的 Storage Write API (gRPC) 設定。不過，如果遇到效能問題，可以設定管道選項來調整這些設定。詳情請參閱 Apache Beam 說明文件中的「調整 Storage Write API (gRPC)」。

串流管道

以下建議適用於串流管道。

如果是串流管道，建議使用 Storage Write API (gRPC) (STORAGE_WRITE_API 或 STORAGE_API_AT_LEAST_ONCE)。
串流管道可以使用檔案載入，但這種做法有缺點：
- 必須使用視窗才能寫入檔案。無法使用全域視窗。
- 使用共用運算單元集區時，BigQuery 會盡可能載入檔案。從記錄寫入到 BigQuery 可用之間，可能會有明顯的延遲。
- 如果載入工作失敗 (例如資料不佳或結構定義不符)，整個管道就會失敗。
建議盡可能使用 STORAGE_WRITE_API_AT_LEAST_ONCE。這可能會導致重複的記錄寫入 BigQuery，但費用比 STORAGE_WRITE_API 低，且擴充性更高。
一般來說，請避免使用 STREAMING_INSERTS。串流資料即時插入的費用高於 Storage Write API (gRPC)，效能也較差，且採用盡力簡化，可能會產生一些重複項目。
資料 sharding 可提升串流管道的效能。對大多數管道來說，自動資料分割是不錯的起點。不過，您可以按照下列方式調整資料分割：
- 如果是 STORAGE_WRITE_API，請呼叫 withNumStorageWriteApiStreams 設定寫入串流數量。
- 如果是 FILE_LOADS，請呼叫 withNumFileShards 設定檔案 shard 數量。
如果您使用串流資料即時插入，建議將 retryTransientErrors 設為重試政策。

批次管道

以下建議適用於批次管道。

對於大多數大型批次管道，建議先嘗試 FILE_LOADS。批次管道可以使用 STORAGE_WRITE_API，但如果規模較大 (超過 1,000 個 vCPU) 或同時執行多個管道，則可能會超出配額限制。Apache Beam 不會限制批次 STORAGE_WRITE_API 工作的寫入串流數量上限，因此工作最終會達到 BigQuery Storage API 限制。
使用 FILE_LOADS 時，共用的 BigQuery 運算單元集區或預留運算單元集區可能會用盡。如果遇到這類失敗，請嘗試下列方法：
- 減少工作站數量上限或工作站大小。
- 購買更多預留運算單元。
- 建議使用 STORAGE_WRITE_API。
中小型管道 (<1,000 個 vCPU) 可能適合使用 STORAGE_WRITE_API。對於這些較小的工作，如果您需要無效信件佇列，或 FILE_LOADS 共用運算單元集區不夠用，請考慮使用 STORAGE_WRITE_API。
如果可以接受重複資料，請考慮使用 STORAGE_WRITE_API_AT_LEAST_ONCE。這個模式可能會導致重複的記錄寫入 BigQuery，但費用可能比 STORAGE_WRITE_API 選項低。
不同的寫入模式可能會因管道特性而有不同的效能。建議您多方嘗試，找出最適合工作負載的寫入模式。

處理資料列層級錯誤

本節說明如何處理資料列層級可能發生的錯誤，例如輸入資料格式不正確或結構定義不相符。

如果是 Storage Write API (gRPC)，任何無法寫入的資料列都會放入獨立的 PCollection。如要取得這個集合，請在 WriteResult 物件上呼叫 getFailedStorageApiInserts。如需這種做法的範例，請參閱「將資料串流至 BigQuery」。

建議您將錯誤傳送至無效信件佇列或資料表，以便日後處理。如要進一步瞭解這個模式，請參閱「BigQueryIO 停滯訊息模式」。

如果是 FILE_LOADS，載入資料時發生錯誤，載入工作就會失敗，管道也會擲回執行階段例外狀況。您可以在 Dataflow 記錄中查看錯誤，或查看 BigQuery 工作記錄。I/O 連接器不會傳回個別失敗資料列的資訊。

如要進一步瞭解如何排解錯誤，請參閱「BigQuery 連接器錯誤」。

範例

下列範例說明如何使用 Dataflow 寫入 BigQuery。這些範例使用 BigQueryIO 連接器。

寫入現有資料表

下列範例會建立批次管道，將 PCollection<MyData> 寫入 BigQuery，其中 MyData 是自訂資料類型。

BigQueryIO.write() 方法會傳回 BigQueryIO.Write<T> 類型，用於設定寫入作業。詳情請參閱 Apache Beam 說明文件中的「寫入資料表」。這個程式碼範例會寫入現有資料表 (CREATE_NEVER)，並將新資料列附加至資料表 (WRITE_APPEND)。

Java

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

import com.google.api.services.bigquery.model.TableRow;
import java.util.Arrays;
import java.util.List;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.coders.DefaultCoder;
import org.apache.beam.sdk.extensions.avro.coders.AvroCoder;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.CreateDisposition;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.WriteDisposition;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;

public class BigQueryWrite {
  // A custom datatype for the source data.
  @DefaultCoder(AvroCoder.class)
  public static class MyData {
    public String name;
    public Long age;

    public MyData() {}

    public MyData(String name, Long age) {
      this.name = name;
      this.age = age;
    }
  }

  public static void main(String[] args) {
    // Example source data.
    final List<MyData> data = Arrays.asList(
        new MyData("Alice", 40L),
        new MyData("Bob", 30L),
        new MyData("Charlie", 20L)
    );

    // Parse the pipeline options passed into the application. Example:
    //   --projectId=$PROJECT_ID --datasetName=$DATASET_NAME --tableName=$TABLE_NAME
    // For more information, see https://beam.apache.org/documentation/programming-guide/#configuring-pipeline-options
    PipelineOptionsFactory.register(ExamplePipelineOptions.class);
    ExamplePipelineOptions options = PipelineOptionsFactory.fromArgs(args)
        .withValidation()
        .as(ExamplePipelineOptions.class);

    // Create a pipeline and apply transforms.
    Pipeline pipeline = Pipeline.create(options);
    pipeline
        // Create an in-memory PCollection of MyData objects.
        .apply(Create.of(data))
        // Write the data to an exiting BigQuery table.
        .apply(BigQueryIO.<MyData>write()
            .to(String.format("%s:%s.%s",
                options.getProjectId(),
                options.getDatasetName(),
                options.getTableName()))
            .withFormatFunction(
                (MyData x) -> new TableRow().set("user_name", x.name).set("age", x.age))
            .withCreateDisposition(CreateDisposition.CREATE_NEVER)
            .withWriteDisposition(WriteDisposition.WRITE_APPEND)
            .withMethod(Write.Method.STORAGE_WRITE_API));
    pipeline.run().waitUntilFinish();
  }
}

寫入新的或現有資料表

在以下範例中，如果目的地資料表不存在，系統會將建立處置設為 CREATE_IF_NEEDED，建立新的資料表。使用這個選項時，必須提供資料表結構定義。如果連接器建立新資料表，就會使用這個結構定義。

Java

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

import com.google.api.services.bigquery.model.TableFieldSchema;
import com.google.api.services.bigquery.model.TableRow;
import com.google.api.services.bigquery.model.TableSchema;
import java.util.Arrays;
import java.util.List;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.coders.DefaultCoder;
import org.apache.beam.sdk.extensions.avro.coders.AvroCoder;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.CreateDisposition;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;

public class BigQueryWriteWithSchema {
  // A custom datatype for the source data.
  @DefaultCoder(AvroCoder.class)
  public static class MyData {
    public String name;
    public Long age;

    public MyData() {}

    public MyData(String name, Long age) {
      this.name = name;
      this.age = age;
    }
  }

  public static void main(String[] args) {
    // Example source data.
    final List<MyData> data = Arrays.asList(
        new MyData("Alice", 40L),
        new MyData("Bob", 30L),
        new MyData("Charlie", 20L)
    );

    // Define a table schema. A schema is required for write disposition CREATE_IF_NEEDED.
    TableSchema schema = new TableSchema()
        .setFields(
            Arrays.asList(
                new TableFieldSchema()
                    .setName("user_name")
                    .setType("STRING")
                    .setMode("REQUIRED"),
                new TableFieldSchema()
                    .setName("age")
                    .setType("INT64") // Defaults to NULLABLE
            )
        );

    // Parse the pipeline options passed into the application. Example:
    //   --projectId=$PROJECT_ID --datasetName=$DATASET_NAME --tableName=$TABLE_NAME
    // For more information, see https://beam.apache.org/documentation/programming-guide/#configuring-pipeline-options
    PipelineOptionsFactory.register(ExamplePipelineOptions.class);
    ExamplePipelineOptions options = PipelineOptionsFactory.fromArgs(args)
        .withValidation()
        .as(ExamplePipelineOptions.class);

    // Create a pipeline and apply transforms.
    Pipeline pipeline = Pipeline.create(options);
    pipeline
        // Create an in-memory PCollection of MyData objects.
        .apply(Create.of(data))
        // Write the data to a new or existing BigQuery table.
        .apply(BigQueryIO.<MyData>write()
            .to(String.format("%s:%s.%s",
                options.getProjectId(),
                options.getDatasetName(),
                options.getTableName()))
            .withFormatFunction(
                (MyData x) -> new TableRow().set("user_name", x.name).set("age", x.age))
            .withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
            .withSchema(schema)
            .withMethod(Write.Method.STORAGE_WRITE_API)
        );
    pipeline.run().waitUntilFinish();
  }
}

將資料串流至 BigQuery

以下範例說明如何將寫入模式設為 STORAGE_WRITE_API，使用僅一次語意串流資料

並非所有串流管道都需要「僅限一次」語意。例如，您可以從目的地資料表手動移除重複項目。如果您的情境可以接受重複記錄，建議您將寫入方法設為 STORAGE_API_AT_LEAST_ONCE，採用至少一次語意。這種方法通常更有效率，且可降低大多數管道的延遲時間。

Java

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

import com.google.api.services.bigquery.model.TableRow;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.CreateDisposition;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.WriteDisposition;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.testing.TestStream;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.values.TimestampedValue;
import org.apache.beam.sdk.values.TypeDescriptor;
import org.apache.beam.sdk.values.TypeDescriptors;
import org.joda.time.Duration;
import org.joda.time.Instant;

public class BigQueryStreamExactlyOnce {
  // Create a PTransform that sends simulated streaming data. In a real application, the data
  // source would be an external source, such as Pub/Sub.
  private static TestStream<String> createEventSource() {
    Instant startTime = new Instant(0);
    return TestStream.create(StringUtf8Coder.of())
        .advanceWatermarkTo(startTime)
        .addElements(
            TimestampedValue.of("Alice,20", startTime),
            TimestampedValue.of("Bob,30",
                startTime.plus(Duration.standardSeconds(1))),
            TimestampedValue.of("Charles,40",
                startTime.plus(Duration.standardSeconds(2))),
            TimestampedValue.of("Dylan,Invalid value",
                startTime.plus(Duration.standardSeconds(2))))
        .advanceWatermarkToInfinity();
  }

  public static PipelineResult main(String[] args) {
    // Parse the pipeline options passed into the application. Example:
    //   --projectId=$PROJECT_ID --datasetName=$DATASET_NAME --tableName=$TABLE_NAME
    // For more information, see https://beam.apache.org/documentation/programming-guide/#configuring-pipeline-options
    PipelineOptionsFactory.register(ExamplePipelineOptions.class);
    ExamplePipelineOptions options = PipelineOptionsFactory.fromArgs(args)
        .withValidation()
        .as(ExamplePipelineOptions.class);
    options.setStreaming(true);

    // Create a pipeline and apply transforms.
    Pipeline pipeline = Pipeline.create(options);
    pipeline
        // Add a streaming data source.
        .apply(createEventSource())
        // Map the event data into TableRow objects.
        .apply(MapElements
            .into(TypeDescriptor.of(TableRow.class))
            .via((String x) -> {
              String[] columns = x.split(",");
              return new TableRow().set("user_name", columns[0]).set("age", columns[1]);
            }))
        // Write the rows to BigQuery
        .apply(BigQueryIO.writeTableRows()
            .to(String.format("%s:%s.%s",
                options.getProjectId(),
                options.getDatasetName(),
                options.getTableName()))
            .withCreateDisposition(CreateDisposition.CREATE_NEVER)
            .withWriteDisposition(WriteDisposition.WRITE_APPEND)
            .withMethod(Write.Method.STORAGE_WRITE_API)
            // For exactly-once processing, set the triggering frequency.
            .withTriggeringFrequency(Duration.standardSeconds(5)))
        // Get the collection of write errors.
        .getFailedStorageApiInserts()
        .apply(MapElements.into(TypeDescriptors.strings())
            // Process each error. In production systems, it's useful to write the errors to
            // another destination, such as a dead-letter table or queue.
            .via(
                x -> {
                  System.out.println("Failed insert: " + x.getErrorMessage());
                  System.out.println("Row: " + x.getRow());
                  return "";
                }));
    return pipeline.run();
  }
}

從 Dataflow 寫入 BigQuery 透過集合功能整理內容 你可以依據偏好儲存及分類內容。

總覽

效能

使用 BigQueryIO 連接器

平行處理工作數量

最佳做法

串流管道

批次管道

處理資料列層級錯誤

範例

寫入現有資料表

Java

寫入新的或現有資料表

Java

將資料串流至 BigQuery

Java

後續步驟

從 Dataflow 寫入 BigQuery

使用 `BigQueryIO` 連接器