Google menggunakan teknologi AI untuk menerjemahkan konten ke dalam bahasa pilihan Anda. Terjemahan AI mungkin mengandung kesalahan.

Menulis dari Dataflow ke BigQuery

Dokumen ini menjelaskan cara menulis data dari Dataflow ke BigQuery.

Ringkasan

Untuk sebagian besar kasus penggunaan, pertimbangkan untuk menggunakan I/O Terkelola untuk menulis ke BigQuery. I/O terkelola menyediakan fitur seperti upgrade otomatis dan API konfigurasi yang konsisten. Saat menulis ke BigQuery, Managed I/O otomatis memilih metode penulisan terbaik untuk tugas batch atau streaming.

Jika Anda memerlukan penyesuaian performa yang lebih canggih, pertimbangkan untuk menggunakan konektor BigQueryIO. Untuk mengetahui informasi selengkapnya, lihat Menggunakan konektor BigQueryIO dalam dokumen ini.

Performa

Tabel berikut menunjukkan metrik performa untuk berbagai workload. Workload ini dijalankan pada satu worker e2-standard2, menggunakan Apache Beam SDK 2.49.0 untuk Java. Mereka tidak menggunakan Runner v2.

100 juta data \| 1 kB \| 1 kolom	Throughput (byte)	Throughput (elemen)
Storage Write	55 MBps	54.000 elemen per detik
Pemuatan Avro	78 MBps	77.000 elemen per detik
Json Load	54 MBps	53.000 elemen per detik

Metrik ini didasarkan pada pipeline batch sederhana. Tujuannya adalah untuk membandingkan performa antara konektor I/O, dan tidak selalu mewakili pipeline di dunia nyata. Performa pipeline Dataflow bersifat kompleks, dan merupakan fungsi dari jenis VM, data yang sedang diproses, performa sumber dan sink eksternal, serta kode pengguna. Metrik didasarkan pada menjalankan Java SDK, dan tidak mewakili karakteristik performa SDK bahasa lain. Untuk mengetahui informasi selengkapnya, lihat Performa IO Beam.

Menggunakan konektor `BigQueryIO`

Konektor I/O BigQuery mendukung metode berikut untuk menulis ke BigQuery:

STORAGE_WRITE_API. Dalam mode ini, konektor melakukan penulisan langsung ke penyimpanan BigQuery, menggunakan BigQuery Storage Write API (gRPC). Storage Write API (gRPC) menggabungkan penyerapan streaming dan pemuatan batch ke dalam satu API berperforma tinggi. Mode ini menjamin semantik tepat satu kali.
STORAGE_API_AT_LEAST_ONCE. Mode ini juga menggunakan Storage Write API (gRPC), tetapi menyediakan semantik setidaknya sekali. Mode ini menghasilkan latensi yang lebih rendah untuk sebagian besar pipeline. Namun, penulisan duplikat mungkin terjadi.
FILE_LOADS. Dalam mode ini, konektor menulis data input ke file staging di Cloud Storage. Kemudian, layanan ini akan menjalankan tugas pemuatan BigQuery untuk memuat data ke BigQuery. Mode ini adalah default untuk PCollections terbatas, yang paling sering ditemukan di pipeline batch.
STREAMING_INSERTS. Dalam mode ini, konektor menggunakan API streaming lama. Mode ini menggunakan penghapusan duplikat dengan upaya terbaik, yang dapat menghasilkan beberapa duplikat. Mode ini adalah default untuk PCollections yang tidak terbatas, tetapi tidak direkomendasikan untuk project baru.

Saat memilih metode penulisan, pertimbangkan poin-poin berikut:

Untuk tugas streaming, pertimbangkan untuk menggunakan STORAGE_WRITE_API atau STORAGE_API_AT_LEAST_ONCE, karena mode ini menulis langsung ke penyimpanan BigQuery, tanpa menggunakan file staging perantara.
Jika Anda menjalankan pipeline menggunakan mode streaming setidaknya sekali, tetapkan mode tulis ke STORAGE_API_AT_LEAST_ONCE. Setelan ini lebih efisien dan sesuai dengan semantik mode streaming setidaknya sekali. Metode penulisan FILE_LOADS tidak kompatibel dengan mode streaming minimal satu kali.
Pemuatan file dan Storage Write API (gRPC) memiliki kuota dan batas yang berbeda.
Tugas pemuatan menggunakan kumpulan slot BigQuery bersama atau slot yang dicadangkan. Untuk menggunakan slot yang direservasi, jalankan tugas pemuatan dalam project dengan penetapan pemesanan jenis PIPELINE. Tugas pemuatan tidak dikenai biaya jika Anda menggunakan kumpulan slot BigQuery bersama. Namun, BigQuery tidak menjamin ketersediaan kapasitas dari gabungan slot bersama. Untuk mengetahui informasi selengkapnya, lihat Pengantar pemesanan.

Keparalelan

Untuk FILE_LOADS dan STORAGE_WRITE_API dalam pipeline streaming, konektor melakukan sharding data ke sejumlah file atau aliran data. Secara umum, sebaiknya panggil withAutoSharding untuk mengaktifkan sharding otomatis.
Untuk FILE_LOADS di pipeline batch, konektor menulis data ke file yang dipartisi, yang kemudian dimuat ke BigQuery secara paralel.
Untuk STORAGE_WRITE_API di pipeline batch, setiap worker membuat satu atau beberapa aliran data untuk ditulis ke BigQuery, yang ditentukan oleh jumlah total shard.
Untuk STORAGE_API_AT_LEAST_ONCE, ada satu aliran tulis default. Beberapa worker menambahkan ke aliran data ini.

Praktik terbaik

Storage Write API (gRPC) memiliki batas kuota. Konektor menangani batas ini untuk sebagian besar pipeline. Namun, beberapa skenario dapat menghabiskan aliran data Storage Write API (gRPC) yang tersedia. Misalnya, masalah ini mungkin terjadi dalam pipeline yang menggunakan sharding otomatis dan penskalaan otomatis dengan sejumlah besar tujuan, terutama dalam tugas yang berjalan lama dengan workload yang sangat bervariasi. Jika masalah ini terjadi, pertimbangkan untuk menggunakan STORAGE_WRITE_API_AT_LEAST_ONCE, yang dapat menghindari masalah tersebut.
Gunakan metrikGoogle Cloud untuk memantau penggunaan kuota Storage Write API (gRPC) Anda.
Saat menggunakan pemuatan file, Avro biasanya mengungguli JSON. Untuk menggunakan Avro, panggil withAvroFormatFunction.
Secara default, tugas pemuatan berjalan di project yang sama dengan tugas Dataflow. Untuk menentukan project lain, panggil withLoadJobProjectId.
Saat menggunakan Java SDK, pertimbangkan untuk membuat class yang mewakili skema tabel BigQuery. Kemudian, panggil useBeamSchema di pipeline Anda untuk otomatis mengonversi antara jenis Row Apache Beam dan TableRow BigQuery. Untuk contoh class skema, lihat ExampleModel.java.
Jika Anda memuat tabel dengan skema kompleks yang berisi ribuan kolom, pertimbangkan untuk memanggil withMaxBytesPerPartition guna menetapkan ukuran maksimum yang lebih kecil untuk setiap tugas pemuatan.
Secara default, BigQueryIO menggunakan setelan Storage Write API (gRPC) yang sesuai untuk sebagian besar pipeline. Namun, jika melihat masalah performa, Anda dapat menetapkan opsi pipeline untuk menyesuaikan setelan ini. Untuk mengetahui informasi selengkapnya, lihat Menyesuaikan Storage Write API (gRPC) dalam dokumentasi Apache Beam.

Pipeline streaming

Rekomendasi berikut berlaku untuk pipeline streaming.

Untuk pipeline streaming, sebaiknya gunakan Storage Write API (gRPC) (STORAGE_WRITE_API atau STORAGE_API_AT_LEAST_ONCE).
Pipeline streaming dapat menggunakan pemuatan file, tetapi pendekatan ini memiliki kelemahan:
- Pipeline ini memerlukan windowing untuk menulis file. Anda tidak dapat menggunakan jendela global.
- BigQuery memuat file berdasarkan upaya terbaik saat menggunakan gabungan slot bersama. Mungkin ada penundaan yang signifikan antara saat data ditulis dan saat data tersedia di BigQuery.
- Jika tugas pemuatan gagal — misalnya, karena data yang buruk atau ketidakcocokan skema — seluruh pipeline akan gagal.
Sebaiknya gunakan STORAGE_WRITE_API_AT_LEAST_ONCE jika memungkinkan. Hal ini dapat menyebabkan data duplikat ditulis ke BigQuery, tetapi lebih murah dan lebih skalabel daripada STORAGE_WRITE_API.
Secara umum, hindari penggunaan STREAMING_INSERTS. Streaming insert lebih mahal daripada Storage Write API (gRPC), tidak berperforma sebaik itu, dan menggunakan deduplikasi upaya terbaik, yang dapat menghasilkan beberapa duplikat.
Sharding data dapat meningkatkan performa dalam pipeline streaming. Untuk sebagian besar pipeline, sharding otomatis adalah titik awal yang baik. Namun, Anda dapat menyesuaikan sharding sebagai berikut:
- Untuk STORAGE_WRITE_API, panggil withNumStorageWriteApiStreams untuk menetapkan jumlah aliran data tulis.
- Untuk FILE_LOADS, panggil withNumFileShards untuk menetapkan jumlah shard file.
Jika Anda menggunakan streaming insert, sebaiknya tetapkan retryTransientErrors sebagai kebijakan percobaan ulang.

Pipeline batch

Rekomendasi berikut berlaku untuk pipeline batch.

Untuk sebagian besar pipeline batch besar, sebaiknya coba FILE_LOADS terlebih dahulu. Pipeline batch dapat menggunakan STORAGE_WRITE_API, tetapi kemungkinan akan melebihi batas kuota dalam skala besar (lebih dari 1.000 vCPU) atau jika pipeline serentak sedang berjalan. Apache Beam tidak membatasi jumlah maksimum aliran tulis untuk tugas STORAGE_WRITE_API batch, sehingga tugas tersebut pada akhirnya mencapai batas BigQuery Storage API.
Saat menggunakan FILE_LOADS, Anda mungkin menghabiskan gabungan slot BigQuery bersama atau gabungan slot yang dipesan. Jika Anda mengalami kegagalan semacam ini, coba pendekatan berikut:
- Kurangi jumlah maksimum worker atau ukuran worker untuk tugas tersebut.
- Membeli lebih banyak slot yang dipesan.
- Sebaiknya gunakan STORAGE_WRITE_API.
Pipeline kecil hingga menengah (<1.000 vCPU) mungkin akan mendapatkan manfaat dari penggunaan STORAGE_WRITE_API. Untuk tugas yang lebih kecil ini, pertimbangkan untuk menggunakan STORAGE_WRITE_API jika Anda menginginkan antrean pesan yang tidak terkirim atau saat gabungan slot bersama FILE_LOADS tidak cukup.
Jika Anda dapat menoleransi data duplikat, pertimbangkan untuk menggunakan STORAGE_WRITE_API_AT_LEAST_ONCE. Mode ini dapat menyebabkan data duplikat ditulis ke BigQuery, tetapi mungkin lebih murah daripada opsi STORAGE_WRITE_API.
Mode tulis yang berbeda mungkin berperforma berbeda berdasarkan karakteristik pipeline Anda. Lakukan eksperimen untuk menemukan mode tulis terbaik untuk workload Anda.

Menangani error tingkat baris

Bagian ini menjelaskan cara menangani error yang mungkin terjadi di tingkat baris, misalnya karena data input yang tidak terbentuk dengan baik atau ketidakcocokan skema.

Untuk Storage Write API (gRPC), baris apa pun yang tidak dapat ditulis akan ditempatkan ke dalam PCollection terpisah. Untuk mendapatkan koleksi ini, panggil getFailedStorageApiInserts pada objek WriteResult. Untuk contoh pendekatan ini, lihat Melakukan streaming data ke BigQuery.

Sebaiknya kirim error ke antrean atau tabel yang tidak aktif, untuk diproses nanti. Untuk mengetahui informasi selengkapnya tentang pola ini, lihat pola pesan yang tidak terkirim BigQueryIO.

Untuk FILE_LOADS, jika terjadi error saat memuat data, tugas pemuatan akan gagal dan pipeline akan menampilkan pengecualian runtime. Anda dapat melihat error di log Dataflow atau melihat histori tugas BigQuery. Konektor I/O tidak menampilkan informasi tentang setiap baris yang gagal.

Untuk mengetahui informasi selengkapnya tentang pemecahan masalah error, lihat Error konektor BigQuery.

Contoh

Contoh berikut menunjukkan cara menggunakan Dataflow untuk menulis ke BigQuery. Contoh ini menggunakan konektor BigQueryIO.

Menulis ke tabel yang ada

Contoh berikut membuat pipeline batch yang menulis PCollection<MyData> ke BigQuery, dengan MyData sebagai jenis data kustom.

Metode BigQueryIO.write() menampilkan jenis BigQueryIO.Write<T>, yang digunakan untuk mengonfigurasi operasi tulis. Untuk mengetahui informasi selengkapnya, lihat Menulis ke tabel dalam dokumentasi Apache Beam. Contoh kode ini menulis ke tabel yang ada (CREATE_NEVER) dan menambahkan baris baru ke tabel (WRITE_APPEND).

Java

Untuk melakukan autentikasi ke Dataflow, siapkan Kredensial Default Aplikasi. Untuk mengetahui informasi selengkapnya, lihat Menyiapkan autentikasi untuk lingkungan pengembangan lokal.

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();
  }
}

Menulis ke tabel baru atau yang sudah ada

Contoh berikut membuat tabel baru jika tabel tujuan tidak ada, dengan menetapkan disposisi pembuatan ke CREATE_IF_NEEDED. Saat menggunakan opsi ini, Anda harus menyediakan skema tabel. Konektor menggunakan skema ini jika membuat tabel baru.

Java

Untuk melakukan autentikasi ke Dataflow, siapkan Kredensial Default Aplikasi. Untuk mengetahui informasi selengkapnya, lihat Menyiapkan autentikasi untuk lingkungan pengembangan lokal.

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();
  }
}

Melakukan streaming data ke BigQuery

Contoh berikut menunjukkan cara melakukan streaming data menggunakan semantik tepat satu kali, dengan menetapkan mode tulis ke STORAGE_WRITE_API

Tidak semua pipeline streaming memerlukan semantik tepat satu kali. Misalnya, Anda mungkin dapat menghapus duplikat secara manual dari tabel tujuan. Jika kemungkinan data duplikat dapat diterima untuk skenario Anda, pertimbangkan untuk menggunakan semantik setidaknya sekali dengan menetapkan metode tulis ke STORAGE_API_AT_LEAST_ONCE. Metode ini umumnya lebih efisien dan menghasilkan latensi yang lebih rendah untuk sebagian besar pipeline.

Java

Untuk melakukan autentikasi ke Dataflow, siapkan Kredensial Default Aplikasi. Untuk mengetahui informasi selengkapnya, lihat Menyiapkan autentikasi untuk lingkungan pengembangan lokal.

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();
  }
}

Langkah berikutnya

Pelajari lebih lanjut I/O Terkelola.
Pelajari praktik terbaik Pub/Sub ke BigQuery.
Memahami karakteristik performa pipeline Pub/Sub ke BigQuery

Menulis dari Dataflow ke BigQuery Tetap teratur dengan koleksi Simpan dan kategorikan konten berdasarkan preferensi Anda.

Ringkasan

Performa

Menggunakan konektor BigQueryIO

Keparalelan

Praktik terbaik

Pipeline streaming

Pipeline batch

Menangani error tingkat baris

Contoh

Menulis ke tabel yang ada

Java

Menulis ke tabel baru atau yang sudah ada

Java

Melakukan streaming data ke BigQuery

Java

Langkah berikutnya

Menulis dari Dataflow ke BigQuery

Menggunakan konektor `BigQueryIO`