Aliran data perubahan Bigtable ke template BigQuery

Aliran data perubahan Bigtable ke template BigQuery adalah pipeline streaming yang mengalirkan kumpulan data perubahan data Bigtable dan menulisnya ke dalam tabel BigQuery menggunakan Dataflow.

Aliran perubahan Bigtable memungkinkan Anda berlangganan mutasi data per tabel. Jika Anda berlangganan aliran data perubahan tabel, batasan berikut akan berlaku:

• Hanya sel yang dimodifikasi dan deskriptor dari operasi penghapusan yang ditampilkan.
• Hanya nilai baru dari sel yang dimodifikasi yang dikembalikan.

Ketika kumpulan data perubahan data ditulis ke BigQuery, baris mungkin disisipkan secara tidak berurutan dibandingkan dengan pengurutan stempel waktu commit Bigtable asli.

Baris tabel log perubahan yang tidak dapat ditulis ke BigQuery karena error persisten akan ditempatkan secara permanen ke dalam direktori antrean yang dihentikan pengirimannya (antrean pesan yang belum diproses) di Cloud Storage untuk ditinjau secara manual atau diproses lebih lanjut oleh pengguna.

Jika tabel BigQuery yang diperlukan tidak ada, pipeline akan membuatnya. Jika tidak, tabel BigQuery yang sudah ada akan digunakan. Skema tabel BigQuery yang ada harus berisi kolom dalam tabel berikut.

Setiap baris BigQuery baru menyertakan satu kumpulan data perubahan data yang ditampilkan oleh aliran perubahan dari baris yang sesuai di tabel Bigtable Anda.

Skema tabel output BigQuery

Nama kolom	Jenis	Nullable	Deskripsi
row_key	STRING atau BYTES	Tidak	Tombol baris untuk baris yang diubah. Jika opsi pipeline writeRowkeyAsBytes ditetapkan ke true, jenis kolom harus BYTES. Jika tidak, gunakan jenis STRING.
mod_type	STRING	Tidak	Jenis mutasi baris. Gunakan salah satu nilai berikut: SET_CELL, DELETE_CELLS, atau DELETE_FAMILY.
column_family	STRING	Tidak	Grup kolom yang terpengaruh oleh mutasi baris.
column	STRING	Ya	Penentu kolom yang terpengaruh oleh mutasi baris. Untuk jenis mutasi DELETE_FAMILY, tetapkan ke NULL.
commit_timestamp	TIMESTAMP	Tidak	Waktu ketika Bigtable menerapkan mutasi.
big_query_commit_timestamp	TIMESTAMP	Ya	Opsional: Menentukan waktu saat BigQuery menulis baris ke tabel output. Kolom ini tidak diisi jika nama kolom ada di nilai opsi pipeline bigQueryChangelogTableFieldsToIgnore.
timestamp	TIMESTAMP atau INT64	Ya	Nilai stempel waktu sel yang terpengaruh oleh mutasi. Jika opsi pipeline writeNumericTimestamps ditetapkan ke true, jenis kolom harus INT64. Jika tidak, gunakan jenis TIMESTAMP. Untuk jenis mutasi DELETE_CELLS dan DELETE_FAMILY, tetapkan ke NULL.
timestamp_from	TIMESTAMP atau INT64	Ya	Menjelaskan awal inklusif interval stempel waktu untuk semua sel yang dihapus oleh mutasi DELETE_CELLS. Untuk jenis mutasi lainnya, tetapkan ke NULL.
timestamp_to	TIMESTAMP atau INT64	Ya	Menjelaskan akhir eksklusif interval stempel waktu untuk semua sel yang dihapus oleh mutasi DELETE_CELLS. Untuk jenis mutasi lainnya, tetapkan ke NULL.
is_gc	BOOL	Tidak	Opsional: Jika mutasi dipicu oleh kebijakan pembersihan sampah memori, tetapkan ke true. Dalam semua kasus lain, tetapkan ke false. Kolom ini tidak diisi saat nama kolom ada di nilai opsi pipeline bigQueryChangelogTableFieldsToIgnore.
source_instance	STRING	Tidak	Opsional: Menjelaskan nama instance Bigtable tempat mutasi berasal. Kolom ini tidak diisi saat nama kolom ada di nilai opsi pipeline bigQueryChangelogTableFieldsToIgnore.
source_cluster	STRING	Tidak	Opsional: Menjelaskan nama cluster Bigtable tempat mutasi berasal. Kolom ini tidak diisi saat nama kolom ada di nilai opsi pipeline bigQueryChangelogTableFieldsToIgnore.
source_table	STRING	Tidak	Opsional: Menjelaskan nama tabel Bigtable tempat mutasi diterapkan. Nilai di kolom ini mungkin berguna jika beberapa tabel Bigtable melakukan perubahan ke tabel BigQuery yang sama. Kolom ini tidak diisi saat nama kolom ada di nilai opsi pipeline bigQueryChangelogTableFieldsToIgnore.
tiebreaker	INT64	Tidak	Opsional: Jika dua mutasi didaftarkan secara bersamaan oleh cluster Bigtable berbeda, mutasi dengan nilai tiebreaker tertinggi akan diterapkan ke tabel sumber. Mutasi dengan nilai tiebreaker yang lebih rendah akan dihapus. Kolom ini tidak diisi saat nama kolom ada di nilai opsi pipeline bigQueryChangelogTableFieldsToIgnore.
value	STRING atau BYTES	Ya	Nilai baru yang ditetapkan oleh mutasi. Jika opsi pipeline writeValuesAsBytes ditetapkan ke true, jenis kolom harus BYTES. Jika tidak, gunakan jenis STRING. Nilai ditetapkan untuk mutasi SET_CELL. Untuk jenis mutasi lainnya, nilai ditetapkan ke NULL.

Persyaratan pipeline

• Instance sumber Bigtable yang ditentukan.
• Tabel sumber Bigtable yang ditentukan. Tabel harus mengaktifkan aliran perubahan data.
• Profil aplikasi Bigtable yang ditentukan.
• Set data tujuan BigQuery yang ditentukan.

Parameter template

Parameter	Deskripsi
bigtableReadInstanceId	ID instance Bigtable sumber.
bigtableReadTableId	ID tabel Bigtable sumber.
bigtableChangeStreamAppProfile	ID profil aplikasi Bigtable. Profil aplikasi harus menggunakan perutean cluster tunggal dan mengizinkan transaksi baris tunggal.
bigQueryDataset	Nama set data tabel BigQuery tujuan.
writeNumericTimestamps	Opsional: Tulis stempel waktu Bigtable sebagai INT64 BigQuery. Jika ditetapkan ke true, nilai akan dituliskan ke kolom INT64. Jika tidak, nilai akan ditulis ke kolom TIMESTAMP. Kolom yang terpengaruh: timestamp, timestamp_from, dan timestamp_to. Nilai defaultnya adalah false. Jika disetel ke true, waktu akan diukur dalam mikrodetik sejak epoch Unix (1 Januari 1970 pada UTC).
writeRowkeyAsBytes	Opsional: Tulis rowkey sebagai BigQuery BYTES. Jika ditetapkan ke true, tombol baris akan ditulis ke kolom BYTES. Jika tidak, rowkey akan ditulis ke kolom STRING. Setelan defaultnya adalah false.
writeValuesAsBytes	Opsional: Tulis nilai sebagai BigQuery BYTES. Jika ditetapkan ke true, nilai akan dituliskan ke kolom BYTES. Jika tidak, nilai akan ditulis ke kolom STRING. Setelan defaultnya adalah false.
bigQueryChangelogTableName	Opsional: Nama tabel BigQuery tujuan. Jika tidak ditentukan, nilai bigtableReadTableId + "_changelog" akan digunakan
bigQueryProjectId	Opsional: Project ID set data BigQuery. Defaultnya adalah project untuk tugas Dataflow.
bigtableReadProjectId	Opsional: Project ID Bigtable. Defaultnya adalah project untuk tugas Dataflow.
bigtableChangeStreamMetadataInstanceId	Opsional: Perubahan Bigtable melakukan streaming ID instance metadata.
bigtableChangeStreamMetadataTableTableId	Opsional: Perubahan Bigtable melakukan streaming ID tabel metadata.
bigtableChangeStreamCharset	Opsional: Perubahan Bigtable mengalirkan nama charset saat membaca nilai dan penentu kolom.
bigtableChangeStreamStartTimestamp	Opsional: Stempel waktu awal, inklusif, yang akan digunakan untuk membaca aliran perubahan. Contoh, 2022-05-05T07:59:59Z. Setelan defaultnya adalah stempel waktu waktu mulai pipeline.
bigtableChangeStreamIgnoreColumnFamilies	Opsional: Daftar yang dipisahkan koma untuk nama grup kolom yang akan diabaikan akan diabaikan.
bigtableChangeStreamIgnoreColumns	Opsional: Daftar yang dipisahkan koma untuk perubahan nama kolom yang akan diabaikan.
bigtableChangeStreamName	Opsional: Nama unik untuk pipeline klien. Memungkinkan Anda melanjutkan pemrosesan dari titik penghentian pipeline yang berjalan sebelumnya. Setelan defaultnya adalah nama yang dibuat secara otomatis. Lihat log tugas Dataflow untuk mengetahui nilai yang digunakan.
bigtableChangeStreamResume	Opsional: Jika ditetapkan ke true, pipeline baru akan melanjutkan pemrosesan dari titik saat pipeline yang sebelumnya berjalan dengan nilai bigtableChangeStreamName yang sama dihentikan. Jika pipeline dengan nilai bigtableChangeStreamName yang diberikan belum pernah berjalan, pipeline baru tidak akan dimulai. Jika ditetapkan ke false, pipeline baru akan dimulai. Jika pipeline dengan nilai bigtableChangeStreamName yang sama telah berjalan untuk sumber tertentu, pipeline baru tidak akan dimulai. Nilai default-nya adalah false.
bigQueryChangelogTableFieldsToIgnore	Opsional: Daftar yang dipisahkan koma untuk kolom log perubahan yang, jika ditentukan, tidak dibuat dan diisi. Gunakan salah satu nilai yang didukung berikut: is_gc, source_instance, source_cluster, source_table, tiebreaker, atau big_query_commit_timestamp. Secara default, semua kolom telah diisi.
bigQueryChangelogTablePartitionExpirationMs	Opsional: Menetapkan waktu habis masa berlaku partisi tabel log perubahan, dalam milidetik. Jika ditetapkan ke true, partisi yang lebih lama dari jumlah milidetik yang ditentukan akan dihapus. Secara default, tidak ada tanggal habis masa berlaku yang ditetapkan.
bigQueryChangelogTablePartitionGranularity	Opsional: Menentukan perincian untuk mempartisi tabel log perubahan. Jika ditetapkan, tabel akan dipartisi. Gunakan salah satu nilai yang didukung berikut: HOUR, DAY, MONTH, atau YEAR. Secara default, tabel tidak dipartisi.
dlqDirectory	Opsional: Direktori untuk antrean yang dihentikan pengirimannya. Kumpulan data yang gagal diproses akan disimpan di direktori ini. Defaultnya adalah direktori di bawah lokasi sementara tugas Dataflow. Pada umumnya, Anda dapat menggunakan jalur default.

Menjalankan template

Konsol

1. Buka halaman Create job from template Dataflow.
Buka Buat tugas dari template
2. Di kolom Job name, masukkan nama pekerjaan yang unik.
3. Opsional: Untuk Endpoint regional, pilih nilai dari menu drop-down. Region default-nya adalah us-central1.

   Untuk daftar region tempat Anda dapat menjalankan tugas Dataflow, lihat Lokasi Dataflow.

4. Dari menu drop-down Dataflow template, pilih the Bigtable change streams to BigQuery template.
5. Di kolom parameter yang disediakan, masukkan parameter value Anda.
6. Klik Run job.

gcloud

Catatan: Untuk menggunakan Google Cloud CLI guna menjalankan template flex, Anda harus memiliki Google Cloud CLI versi 284.0.0 atau yang lebih baru.

Di shell atau terminal Anda, jalankan template:

gcloud dataflow flex-template run JOB_NAME \
    --region=REGION_NAME \
    --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/Bigtable_Change_Streams_to_BigQuery \
    --parameters \
bigtableReadInstanceId=BIGTABLE_INSTANCE_ID,\
bigtableReadTableId=BIGTABLE_TABLE_ID,\
bigtableChangeStreamAppProfile=BIGTABLE_APPLICATION_PROFILE_ID,\
bigQueryDataset=BIGQUERY_DESTINATION_DATASET

Ganti kode berikut:

• PROJECT_ID: ID project Google Cloud tempat Anda ingin menjalankan tugas Dataflow
• JOB_NAME: nama pekerjaan unik pilihan Anda
• VERSION: versi template yang ingin Anda gunakan

  Anda dapat menggunakan nilai berikut:

  • latest untuk menggunakan versi terbaru template, yang tersedia di folder induk tidak bertanggal di bucket— gs://dataflow-templates-REGION_NAME/latest/
  • nama versi, seperti 2023-09-12-00_RC00, untuk menggunakan versi template tertentu, yang dapat ditemukan bertingkat di folder induk bertanggal masing-masing dalam bucket— gs://dataflow-templates-REGION_NAME/
  Perhatian: Versi terbaru template mungkin diperbarui dengan perubahan yang dapat menyebabkan gangguan. Lingkungan produksi Anda harus menggunakan template yang disimpan di folder induk bertanggal terbaru agar perubahan yang dapat menyebabkan gangguan ini tidak memengaruhi alur kerja produksi Anda.
• REGION_NAME: region tempat Anda ingin men-deploy tugas Dataflow, misalnya us-central1
• BIGTABLE_INSTANCE_ID: ID instance Bigtable Anda.
• BIGTABLE_TABLE_ID: ID tabel Bigtable Anda.
• BIGTABLE_APPLICATION_PROFILE_ID: ID profil aplikasi Bigtable Anda.
• BIGQUERY_DESTINATION_DATASET: nama set data tujuan BigQuery

API

Untuk menjalankan template menggunakan REST API, kirim permintaan HTTP POST. Untuk informasi selengkapnya tentang API dan cakupan otorisasinya, lihat projects.templates.launch.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch
{
  "launch_parameter": {
    "jobName": "JOB_NAME",
    "containerSpecGcsPath": "gs://dataflow-templates-REGION_NAME/VERSION/flex/Bigtable_Change_Streams_to_BigQuery",
    "parameters": {
        "bigtableReadInstanceId": "BIGTABLE_INSTANCE_ID",
        "bigtableReadTableId": "BIGTABLE_TABLE_ID",
        "bigtableChangeStreamAppProfile": "BIGTABLE_APPLICATION_PROFILE_ID",
        "bigQueryDataset": "BIGQUERY_DESTINATION_DATASET"
    }
  }
}

Ganti kode berikut:

• PROJECT_ID: ID project Google Cloud tempat Anda ingin menjalankan tugas Dataflow
• JOB_NAME: nama pekerjaan unik pilihan Anda
• VERSION: versi template yang ingin Anda gunakan

  Anda dapat menggunakan nilai berikut:

  • latest untuk menggunakan versi terbaru template, yang tersedia di folder induk tidak bertanggal di bucket— gs://dataflow-templates-REGION_NAME/latest/
  • nama versi, seperti 2023-09-12-00_RC00, untuk menggunakan versi template tertentu, yang dapat ditemukan bertingkat di folder induk bertanggal masing-masing dalam bucket— gs://dataflow-templates-REGION_NAME/
  Perhatian: Versi terbaru template mungkin diperbarui dengan perubahan yang dapat menyebabkan gangguan. Lingkungan produksi Anda harus menggunakan template yang disimpan di folder induk bertanggal terbaru agar perubahan yang dapat menyebabkan gangguan ini tidak memengaruhi alur kerja produksi Anda.
• LOCATION: region tempat Anda ingin men-deploy tugas Dataflow, misalnya us-central1
• BIGTABLE_INSTANCE_ID: ID instance Bigtable Anda.
• BIGTABLE_TABLE_ID: ID tabel Bigtable Anda.
• BIGTABLE_APPLICATION_PROFILE_ID: ID profil aplikasi Bigtable Anda.
• BIGQUERY_DESTINATION_DATASET: nama set data tujuan BigQuery

Kode sumber template

Java

/*
 * Copyright (C) 2023 Google LLC
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package com.google.cloud.teleport.v2.templates.bigtablechangestreamstobigquery;

import com.google.api.services.bigquery.model.TableRow;
import com.google.cloud.Timestamp;
import com.google.cloud.bigquery.BigQuery;
import com.google.cloud.bigquery.BigQueryOptions;
import com.google.cloud.bigquery.DatasetId;
import com.google.cloud.bigtable.data.v2.models.ChangeStreamMutation;
import com.google.cloud.bigtable.data.v2.models.DeleteCells;
import com.google.cloud.bigtable.data.v2.models.DeleteFamily;
import com.google.cloud.bigtable.data.v2.models.Entry;
import com.google.cloud.bigtable.data.v2.models.SetCell;
import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.v2.bigtable.options.BigtableCommonOptions.ReadChangeStreamOptions;
import com.google.cloud.teleport.v2.bigtable.options.BigtableCommonOptions.ReadOptions;
import com.google.cloud.teleport.v2.bigtable.utils.UnsupportedEntryException;
import com.google.cloud.teleport.v2.cdc.dlq.DeadLetterQueueManager;
import com.google.cloud.teleport.v2.options.BigtableChangeStreamToBigQueryOptions;
import com.google.cloud.teleport.v2.templates.bigtablechangestreamstobigquery.model.BigQueryDestination;
import com.google.cloud.teleport.v2.templates.bigtablechangestreamstobigquery.model.BigtableSource;
import com.google.cloud.teleport.v2.templates.bigtablechangestreamstobigquery.model.Mod;
import com.google.cloud.teleport.v2.templates.bigtablechangestreamstobigquery.model.ModType;
import com.google.cloud.teleport.v2.templates.bigtablechangestreamstobigquery.schemautils.BigQueryUtils;
import com.google.cloud.teleport.v2.transforms.DLQWriteTransform;
import java.util.ArrayList;
import java.util.List;
import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
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.io.gcp.bigquery.InsertRetryPolicy;
import org.apache.beam.sdk.io.gcp.bigquery.WriteResult;
import org.apache.beam.sdk.io.gcp.bigtable.BigtableIO;
import org.apache.beam.sdk.io.gcp.bigtable.BigtableIO.ExistingPipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.Values;
import org.apache.beam.sdk.values.PCollection;
import org.apache.commons.lang3.StringUtils;
import org.joda.time.Instant;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * This pipeline ingests {@link ChangeStreamMutation} from Bigtable change stream. The {@link
 * ChangeStreamMutation} is then broken into {@link Mod}, which converted into {@link TableRow} and
 * inserted into BigQuery table.
 */
@Template(
    name = "Bigtable_Change_Streams_to_BigQuery",
    category = TemplateCategory.STREAMING,
    displayName = "Cloud Bigtable Change Streams to BigQuery",
    description =
        "Streaming pipeline. Streams Bigtable data change records and writes them into BigQuery using Dataflow Runner V2.",
    optionsClass = BigtableChangeStreamToBigQueryOptions.class,
    optionsOrder = {
      BigtableChangeStreamToBigQueryOptions.class,
      ReadChangeStreamOptions.class,
      ReadOptions.class
    },
    skipOptions = {
      "bigtableReadAppProfile",
      "bigtableAdditionalRetryCodes",
      "bigtableRpcAttemptTimeoutMs",
      "bigtableRpcTimeoutMs"
    },
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/cloud-bigtable-change-streams-to-bigquery",
    flexContainerName = "bigtable-changestreams-to-bigquery",
    contactInformation = "https://cloud.google.com/support",
    streaming = true)
public final class BigtableChangeStreamsToBigQuery {
  private static final Logger LOG = LoggerFactory.getLogger(BigtableChangeStreamsToBigQuery.class);

  private static final String USE_RUNNER_V2_EXPERIMENT = "use_runner_v2";

  /**
   * Main entry point for executing the pipeline.
   *
   * @param args The command-line arguments to the pipeline.
   */
  public static void main(String[] args) {
    LOG.info("Starting to replicate change records from Cloud Bigtable change streams to BigQuery");

    BigtableChangeStreamToBigQueryOptions options =
        PipelineOptionsFactory.fromArgs(args)
            .withValidation()
            .as(BigtableChangeStreamToBigQueryOptions.class);

    run(options);
  }

  private static void setOptions(BigtableChangeStreamToBigQueryOptions options) {
    options.setStreaming(true);
    options.setEnableStreamingEngine(true);

    // Add use_runner_v2 to the experiments option, since change streams connector is only supported
    // on Dataflow runner v2.
    List<String> experiments = options.getExperiments();
    if (experiments == null) {
      experiments = new ArrayList<>();
    }
    boolean hasUseRunnerV2 = false;
    for (String experiment : experiments) {
      if (experiment.equalsIgnoreCase(USE_RUNNER_V2_EXPERIMENT)) {
        hasUseRunnerV2 = true;
        break;
      }
    }
    if (!hasUseRunnerV2) {
      experiments.add(USE_RUNNER_V2_EXPERIMENT);
    }
    options.setExperiments(experiments);
  }

  /**
   * Runs the pipeline with the supplied options.
   *
   * @param options The execution parameters to the pipeline.
   * @return The result of the pipeline execution.
   */
  public static PipelineResult run(BigtableChangeStreamToBigQueryOptions options) {
    setOptions(options);

    String changelogTableName = getBigQueryChangelogTableName(options);
    String bigtableProject = getBigtableProjectId(options);
    String bigQueryProject = getBigQueryProjectId(options);
    String bigQueryDataset = options.getBigQueryDataset();

    // If dataset doesn't exist and not checked, pipeline will start failing only after it sees the
    // first change from Cloud Bigtable. BigQueryIO can create table if it doesn't exist, but it
    // cannot create a dataset
    validateBigQueryDatasetExists(bigQueryProject, bigQueryDataset);

    // Retrieve and parse the startTimestamp
    Instant startTimestamp =
        options.getBigtableChangeStreamStartTimestamp().isEmpty()
            ? Instant.now()
            : toInstant(Timestamp.parseTimestamp(options.getBigtableChangeStreamStartTimestamp()));

    BigtableSource sourceInfo =
        new BigtableSource(
            options.getBigtableReadInstanceId(),
            options.getBigtableReadTableId(),
            getBigtableCharset(options),
            options.getBigtableChangeStreamIgnoreColumnFamilies(),
            options.getBigtableChangeStreamIgnoreColumns(),
            startTimestamp);

    BigQueryDestination destinationInfo =
        new BigQueryDestination(
            bigQueryProject,
            bigQueryDataset,
            changelogTableName,
            options.getWriteRowkeyAsBytes(),
            options.getWriteValuesAsBytes(),
            options.getWriteNumericTimestamps(),
            options.getBigQueryChangelogTablePartitionGranularity(),
            options.getBigQueryChangelogTablePartitionExpirationMs(),
            options.getBigQueryChangelogTableFieldsToIgnore());

    BigQueryUtils bigQuery = new BigQueryUtils(sourceInfo, destinationInfo);

    Pipeline pipeline = Pipeline.create(options);
    DeadLetterQueueManager dlqManager = buildDlqManager(options);

    BigtableIO.ReadChangeStream readChangeStream =
        BigtableIO.readChangeStream()
            .withChangeStreamName(options.getBigtableChangeStreamName())
            .withExistingPipelineOptions(
                options.getBigtableChangeStreamResume()
                    ? ExistingPipelineOptions.RESUME_OR_FAIL
                    : ExistingPipelineOptions.FAIL_IF_EXISTS)
            .withProjectId(bigtableProject)
            .withMetadataTableInstanceId(options.getBigtableChangeStreamMetadataInstanceId())
            .withInstanceId(options.getBigtableReadInstanceId())
            .withTableId(options.getBigtableReadTableId())
            .withAppProfileId(options.getBigtableChangeStreamAppProfile())
            .withStartTime(startTimestamp);

    if (!StringUtils.isBlank(options.getBigtableChangeStreamMetadataTableTableId())) {
      readChangeStream =
          readChangeStream.withMetadataTableTableId(
              options.getBigtableChangeStreamMetadataTableTableId());
    }

    PCollection<ChangeStreamMutation> dataChangeRecord =
        pipeline
            .apply("Read from Cloud Bigtable Change Streams", readChangeStream)
            .apply(Values.create());

    PCollection<TableRow> changeStreamMutationToTableRow =
        dataChangeRecord.apply(
            "ChangeStreamMutation To TableRow",
            ParDo.of(new ChangeStreamMutationToTableRowFn(sourceInfo, bigQuery)));

    Write<TableRow> bigQueryWrite =
        BigQueryIO.<TableRow>write()
            .to(destinationInfo.getBigQueryTableReference())
            .withSchema(bigQuery.getDestinationTableSchema())
            .withFormatFunction(element -> element)
            .withFormatRecordOnFailureFunction(element -> element)
            .withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
            .withWriteDisposition(WriteDisposition.WRITE_APPEND)
            .withExtendedErrorInfo()
            .withMethod(Write.Method.STORAGE_API_AT_LEAST_ONCE)
            .withNumStorageWriteApiStreams(0)
            .withFailedInsertRetryPolicy(InsertRetryPolicy.retryTransientErrors());

    if (destinationInfo.isPartitioned()) {
      bigQueryWrite = bigQueryWrite.withTimePartitioning(bigQuery.getTimePartitioning());
    }

    // Unfortunately, due to https://github.com/apache/beam/issues/24090, it is no longer possible
    // to pass metadata via fake columns when writing to BigQuery. Previously we'd pass something
    // like retry count and then format it out before writing, but BQ would return original object
    // which would allow us to increment retry count and store it to DLQ with incremented number.
    // Because WRITE API doesn't allow access to original object, all metadata values are stripped
    // and we can only rely on retry policy and put all other persistently failing rows to DLQ as
    // a non-retriable severe failure.
    //
    // Since we're not going to be retrying such failures, we'll not use any reading from DLQ
    // capability.

    WriteResult writeResult =
        changeStreamMutationToTableRow.apply("Write To BigQuery", bigQueryWrite);

    writeResult
        .getFailedStorageApiInserts()
        .apply(
            "Failed Mod JSON During BigQuery Writes",
            MapElements.via(new BigQueryDeadLetterQueueSanitizer()))
        .apply(
            "Write rejected TableRow JSON To DLQ",
            DLQWriteTransform.WriteDLQ.newBuilder()
                .withDlqDirectory(dlqManager.getSevereDlqDirectory() + "YYYY/MM/dd/HH/mm/")
                .withTmpDirectory(dlqManager.getSevereDlqDirectory() + "tmp/")
                .setIncludePaneInfo(true)
                .build());

    return pipeline.run();
  }

  private static void validateBigQueryDatasetExists(
      String bigQueryProject, String bigQueryDataset) {
    BigQueryOptions options = BigQueryOptions.newBuilder().build();
    options.setThrowNotFound(true);

    BigQuery bigQuery = options.getService();
    bigQuery.getDataset(DatasetId.of(bigQueryProject, bigQueryDataset));
  }

  private static Instant toInstant(Timestamp timestamp) {
    if (timestamp == null) {
      return null;
    } else {
      return Instant.ofEpochMilli(timestamp.getSeconds() * 1000 + timestamp.getNanos() / 1000000);
    }
  }

  private static DeadLetterQueueManager buildDlqManager(
      BigtableChangeStreamToBigQueryOptions options) {
    String tempLocation =
        options.as(DataflowPipelineOptions.class).getTempLocation().endsWith("/")
            ? options.as(DataflowPipelineOptions.class).getTempLocation()
            : options.as(DataflowPipelineOptions.class).getTempLocation() + "/";
    String dlqDirectory =
        options.getDlqDirectory().isEmpty() ? tempLocation + "dlq/" : options.getDlqDirectory();

    LOG.info("Dead letter queue directory: {}", dlqDirectory);
    return DeadLetterQueueManager.create(dlqDirectory, 1);
  }

  private static String getBigtableCharset(BigtableChangeStreamToBigQueryOptions options) {
    return StringUtils.isEmpty(options.getBigtableChangeStreamCharset())
        ? "UTF-8"
        : options.getBigtableChangeStreamCharset();
  }

  private static String getBigtableProjectId(BigtableChangeStreamToBigQueryOptions options) {
    return StringUtils.isEmpty(options.getBigtableReadProjectId())
        ? options.getProject()
        : options.getBigtableReadProjectId();
  }

  private static String getBigQueryChangelogTableName(
      BigtableChangeStreamToBigQueryOptions options) {
    return StringUtils.isEmpty(options.getBigQueryChangelogTableName())
        ? options.getBigtableReadTableId() + "_changelog"
        : options.getBigQueryChangelogTableName();
  }

  private static String getBigQueryProjectId(BigtableChangeStreamToBigQueryOptions options) {
    return StringUtils.isEmpty(options.getBigQueryProjectId())
        ? options.getProject()
        : options.getBigQueryProjectId();
  }

  /**
   * DoFn that converts a {@link ChangeStreamMutation} to multiple {@link Mod} in serialized JSON
   * format.
   */
  static class ChangeStreamMutationToTableRowFn extends DoFn<ChangeStreamMutation, TableRow> {
    private final BigtableSource sourceInfo;
    private final BigQueryUtils bigQuery;

    ChangeStreamMutationToTableRowFn(BigtableSource source, BigQueryUtils bigQuery) {
      this.sourceInfo = source;
      this.bigQuery = bigQuery;
    }

    @ProcessElement
    public void process(@Element ChangeStreamMutation input, OutputReceiver<TableRow> receiver)
        throws Exception {
      for (Entry entry : input.getEntries()) {
        ModType modType = getModType(entry);

        Mod mod = null;
        switch (modType) {
          case SET_CELL:
            mod = new Mod(sourceInfo, input, (SetCell) entry);
            break;
          case DELETE_CELLS:
            mod = new Mod(sourceInfo, input, (DeleteCells) entry);
            break;
          case DELETE_FAMILY:
            mod = new Mod(sourceInfo, input, (DeleteFamily) entry);
            break;
          default:
          case UNKNOWN:
            throw new UnsupportedEntryException(
                "Cloud Bigtable change stream entry of type "
                    + entry.getClass().getName()
                    + " is not supported. The entry was put into a dead letter queue directory. "
                    + "Please update your Dataflow template with the latest template version");
        }

        TableRow tableRow = new TableRow();
        if (bigQuery.setTableRowFields(mod, tableRow)) {
          receiver.output(tableRow);
        }
      }
    }

    private ModType getModType(Entry entry) {
      if (entry instanceof SetCell) {
        return ModType.SET_CELL;
      } else if (entry instanceof DeleteCells) {
        return ModType.DELETE_CELLS;
      } else if (entry instanceof DeleteFamily) {
        return ModType.DELETE_FAMILY;
      }
      return ModType.UNKNOWN;
    }
  }
}

Langkah selanjutnya

• Pelajari Template Dataflow.
• Lihat daftar template yang disediakan Google.