Membaca dari BigQuery ke Dataflow

Dokumen ini menjelaskan cara membaca data dari BigQuery ke Dataflow.

Ringkasan

Untuk sebagian besar kasus penggunaan, pertimbangkan untuk menggunakan Managed I/O untuk membaca dari BigQuery. I/O Terkelola menyediakan fitur seperti upgrade otomatis dan API konfigurasi yang konsisten. Saat membaca dari BigQuery, I/O Terkelola melakukan pembacaan tabel langsung yang menawarkan performa pembacaan terbaik.

Jika Anda memerlukan penyesuaian performa yang lebih canggih, pertimbangkan untuk menggunakan konektor BigQueryIO. Konektor BigQueryIO mendukung pembacaan tabel langsung dan pembacaan dari tugas ekspor BigQuery. API ini juga menawarkan kontrol yang lebih terperinci atas deserialisasi data tabel. Untuk informasi selengkapnya, lihat Menggunakan konektor BigQueryIO dalam dokumen ini.

Proyeksi dan pemfilteran kolom

Untuk mengurangi volume data yang dibaca pipeline dari BigQuery, Anda dapat menggunakan teknik berikut:

• Proyeksi kolom menentukan subset kolom yang akan dibaca dari tabel. Gunakan proyeksi kolom jika tabel Anda memiliki banyak kolom dan Anda hanya perlu membaca sebagian darinya.
• Pemfilteran baris menentukan predikat yang akan diterapkan ke tabel. Operasi baca BigQuery hanya menampilkan baris yang cocok dengan filter, yang dapat mengurangi jumlah total data yang diserap oleh pipeline.

Contoh berikut membaca kolom "user_name" dan "age" dari tabel dan memfilter baris yang tidak cocok dengan predikat "age > 18". Contoh ini menggunakan I/O Terkelola.

Java

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

import com.google.common.collect.ImmutableMap;
import java.util.List;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.managed.Managed;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.values.Row;
import org.apache.beam.sdk.values.TypeDescriptors;

public class BigQueryReadWithProjectionAndFiltering {
  public static void 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);

    String tableSpec = String.format("%s:%s.%s",
        options.getProjectId(),
        options.getDatasetName(),
        options.getTableName());

    ImmutableMap<String, Object> config = ImmutableMap.<String, Object>builder()
        .put("table", tableSpec)
        .put("row_restriction", "age > 18")
        .put("fields", List.of("user_name", "age"))
        .build();

    // Create a pipeline and apply transforms.
    Pipeline pipeline = Pipeline.create(options);
    pipeline
        .apply(Managed.read(Managed.BIGQUERY).withConfig(config)).getSinglePCollection()
        .apply(MapElements
            .into(TypeDescriptors.strings())
            // Access individual fields in the row.
            .via((Row row) -> {
              String output = String.format("Name: %s, Age: %s%n",
                  row.getString("user_name"),
                  row.getInt64("age"));
              System.out.println(output);
              return output;
            }));
    pipeline.run().waitUntilFinish();
  }
}

Membaca dari hasil kueri

Contoh berikut menggunakan I/O Terkelola untuk membaca hasil kueri SQL. Fungsi ini menjalankan kueri terhadap set data publik BigQuery. Anda juga dapat menggunakan kueri SQL untuk membaca dari tampilan BigQuery atau tampilan terwujud.

Java

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

import com.google.common.collect.ImmutableMap;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.managed.Managed;
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.Row;
import org.apache.beam.sdk.values.TypeDescriptors;

public class BigQueryReadFromQuery {
  public static void main(String[] args) {
    // The SQL query to run inside BigQuery.
    final String queryString =
        "SELECT repo_name as repo, COUNT(*) as count "
            + "FROM `bigquery-public-data.github_repos.sample_commits` "
            + "GROUP BY repo_name";

    // Parse the pipeline options passed into the application.
    // For more information, see https://beam.apache.org/documentation/programming-guide/#configuring-pipeline-options
    PipelineOptions options = PipelineOptionsFactory.fromArgs(args)
        .withValidation().create();

    ImmutableMap<String, Object> config = ImmutableMap.<String, Object>builder()
        .put("query", queryString)
        .build();

    // Create a pipeline and apply transforms.
    Pipeline pipeline = Pipeline.create(options);
    pipeline
        .apply(Managed.read(Managed.BIGQUERY).withConfig(config)).getSinglePCollection()
        .apply(MapElements
            .into(TypeDescriptors.strings())
            // Access individual fields in the row.
            .via((Row row) -> {
              String output = String.format("Repo: %s, commits: %d%n",
                  row.getString("repo"),
                  row.getInt64("count"));
              System.out.println(output);
              return output;
            }));
    pipeline.run().waitUntilFinish();
  }
}

Menggunakan konektor BigQueryIO

Konektor BigQueryIO mendukung metode serialisasi berikut:

• Baca data sebagai data berformat Avro. Dengan menggunakan metode ini, Anda menyediakan fungsi yang mengurai data Avro menjadi jenis data kustom.
• Baca data sebagai objek TableRow. Metode ini mudah karena tidak memerlukan jenis data kustom. Namun, umumnya memiliki performa yang lebih rendah daripada membaca kumpulan data berformat Avro.

Konektor mendukung dua opsi untuk membaca data:

• Ekspor tugas. Secara default, konektor BigQueryIO menjalankan tugas ekspor BigQuery yang menulis data tabel ke Cloud Storage. Konektor kemudian membaca data dari Cloud Storage.
• Pembacaan tabel langsung. Opsi ini lebih cepat daripada tugas ekspor, karena menggunakan BigQuery Storage Read API dan melewati langkah ekspor. Untuk menggunakan pembacaan tabel langsung, panggil withMethod(Method.DIRECT_READ) saat Anda mem-build pipeline.

Saat memilih opsi yang akan digunakan, pertimbangkan poin-poin berikut:

• Secara umum, sebaiknya gunakan pembacaan tabel langsung. Storage Read API lebih cocok untuk pipeline data daripada tugas ekspor, karena tidak memerlukan langkah perantara untuk mengekspor data.

• Jika menggunakan pembacaan langsung, Anda akan ditagih untuk penggunaan Storage Read API. Lihat Harga ekstraksi data di halaman harga BigQuery.

• Tidak ada biaya tambahan untuk tugas ekspor. Namun, tugas ekspor memiliki batas. Untuk pemindahan data dalam jumlah besar, dengan ketepatan waktu sebagai prioritas dan biaya yang dapat disesuaikan, sebaiknya gunakan operasi baca langsung.

• Storage Read API memiliki batas kuota. Gunakan metrikGoogle Cloud untuk memantau penggunaan kuota Anda.

• Jika Anda menggunakan tugas ekspor, tetapkan opsi pipeline --tempLocation untuk menentukan bucket Cloud Storage untuk file yang diekspor.

• Saat menggunakan Storage Read API, Anda mungkin melihat error masa berlaku sewa dan waktu tunggu sesi di log, seperti:

  • DEADLINE_EXCEEDED
  • Server Unresponsive
  • StatusCode.FAILED_PRECONDITION details = "there was an error operating on 'projects/<projectID>/locations/<location>/sessions/<sessionID>/streams/<streamID>': session`

  Error ini dapat terjadi jika operasi memerlukan waktu lebih lama dari waktu tunggu, biasanya dalam pipeline yang berjalan selama lebih dari 6 jam. Untuk mengurangi masalah ini, beralihlah ke ekspor file.

• Tingkat paralelisme bergantung pada metode baca:

  • Pembacaan langsung: Konektor I/O menghasilkan jumlah aliran data yang dinamis, berdasarkan ukuran permintaan ekspor. Fungsi ini membaca streaming ini langsung dari BigQuery secara paralel.

  • Tugas ekspor: BigQuery menentukan jumlah file yang akan ditulis ke Cloud Storage. Jumlah file bergantung pada kueri dan volume data. Konektor I/O membaca file yang diekspor secara paralel.

Tabel berikut menunjukkan metrik performa untuk berbagai opsi pembacaan I/O BigQuery. Beban kerja dijalankan di satu pekerja 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)
Pembacaan Penyimpanan	120 MBps	88.000 elemen per detik
Avro Export	105 MBps	78.000 elemen per detik
Ekspor JSON	110 MBps	81.000 elemen per detik

Metrik ini didasarkan pada pipeline batch sederhana. Pengujian ini dimaksudkan 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 lainnya. Untuk mengetahui informasi selengkapnya, lihat Performa Beam IO.

Contoh

Contoh kode berikut menggunakan konektor BigQueryIO dengan pembacaan tabel langsung. Untuk menggunakan tugas ekspor, hapus panggilan ke withMethod.

Membaca data berformat Avro

Contoh ini menunjukkan cara menggunakan konektor BigQueryIO untuk membaca data berformat Avro.

Untuk membaca data BigQuery ke dalam data berformat Avro, gunakan metode read(SerializableFunction). Metode ini menggunakan fungsi yang ditentukan aplikasi yang mengurai objek SchemaAndRecord dan menampilkan jenis data kustom. Output dari konektor adalah PCollection dari jenis data kustom Anda.

Kode berikut membaca PCollection<MyData> dari tabel BigQuery, dengan MyData adalah class yang ditentukan aplikasi.

Java

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

import org.apache.avro.generic.GenericRecord;
import org.apache.avro.util.Utf8;
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.TypedRead;
import org.apache.beam.sdk.io.gcp.bigquery.SchemaAndRecord;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.SerializableFunction;
import org.apache.beam.sdk.values.TypeDescriptor;

public class BigQueryReadAvro {

  // A custom datatype to hold a record from the source table.
  @DefaultCoder(AvroCoder.class)
  public static class MyData {
    public String name;
    public Long age;

    // Function to convert Avro records to MyData instances.
    public static class FromSchemaAndRecord
            implements SerializableFunction<SchemaAndRecord, MyData> {
      @Override public MyData apply(SchemaAndRecord elem) {
        MyData data = new MyData();
        GenericRecord record = elem.getRecord();
        data.name = ((Utf8) record.get("user_name")).toString();
        data.age = (Long) record.get("age");
        return data;
      }
    }
  }

  public static void 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);

    // Create a pipeline and apply transforms.
    Pipeline pipeline = Pipeline.create(options);
    pipeline
        // Read table data into Avro records, using an application-defined parsing function.
        .apply(BigQueryIO.read(new MyData.FromSchemaAndRecord())
            .from(String.format("%s:%s.%s",
                options.getProjectId(),
                options.getDatasetName(),
                options.getTableName()))
            .withMethod(TypedRead.Method.DIRECT_READ))
        // The output from the previous step is a PCollection<MyData>.
        .apply(MapElements
            .into(TypeDescriptor.of(MyData.class))
            .via((MyData x) -> {
              System.out.printf("Name: %s, Age: %d%n", x.name, x.age);
              return x;
            }));
    pipeline.run().waitUntilFinish();
  }
}

Metode read menggunakan antarmuka SerializableFunction<SchemaAndRecord, T>, yang menentukan fungsi untuk mengonversi dari data Avro ke class data kustom. Dalam contoh kode sebelumnya, metode MyData.apply menerapkan fungsi konversi ini. Contoh fungsi mengurai kolom name dan age dari data Avro dan menampilkan instance MyData.

Catatan: Kode ini gagal jika fungsi penguraian tidak cocok dengan skema sebenarnya dari tabel BigQuery. Secara khusus, kolom yang dinamai harus ada, dan jenis data harus kompatibel. Selain itu, jika ada kolom dalam tabel yang nullable, kode Anda harus menangani nilai null.

Untuk menentukan tabel BigQuery yang akan dibaca, panggil metode from, seperti yang ditunjukkan dalam contoh sebelumnya. Untuk mengetahui informasi selengkapnya, lihat Nama tabel dalam dokumentasi konektor I/O BigQuery.

Membaca objek TableRow

Contoh ini menunjukkan cara menggunakan konektor BigQueryIO untuk membaca objek TableRow.

Metode readTableRows membaca data BigQuery ke dalam PCollection dari objek TableRow. Setiap TableRow adalah peta key-value pair yang menyimpan satu baris data tabel. Tentukan tabel BigQuery yang akan dibaca dengan memanggil metode from.

Kode berikut membaca PCollection<TableRows> dari tabel BigQuery.

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.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.TypedRead.Method;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.values.TypeDescriptor;

public class BiqQueryReadTableRows {
  public static void 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);

    // Create a pipeline and apply transforms.
    Pipeline pipeline = Pipeline.create(options);
    pipeline
        // Read table data into TableRow objects.
        .apply(BigQueryIO.readTableRows()
            .from(String.format("%s:%s.%s",
                options.getProjectId(),
                options.getDatasetName(),
                options.getTableName()))
            .withMethod(Method.DIRECT_READ)
        )
        // The output from the previous step is a PCollection<TableRow>.
        .apply(MapElements
            .into(TypeDescriptor.of(TableRow.class))
            // Use TableRow to access individual fields in the row.
            .via((TableRow row) -> {
              var name = (String) row.get("user_name");
              var age = (String) row.get("age");
              System.out.printf("Name: %s, Age: %s%n", name, age);
              return row;
            }));
    pipeline.run().waitUntilFinish();
  }
}

Contoh ini juga menunjukkan cara mengakses nilai dari kamus TableRow. Nilai bilangan bulat dienkode sebagai string agar cocok dengan format JSON yang diekspor BigQuery.

Catatan: Kode ini gagal jika fungsi penguraian tidak cocok dengan skema sebenarnya dari tabel BigQuery. Secara khusus, kolom yang dinamai harus ada, dan jenis data harus kompatibel. Selain itu, jika ada kolom dalam tabel yang nullable, kode Anda harus menangani nilai null.

Langkah berikutnya

• I/O Terkelola untuk BigQuery
• Menulis ke BigQuery