Template Bigtable change stream ke Pub/Sub

Template Bigtable change streams to Pub/Sub adalah pipeline streaming yang melakukan streaming data Bigtable dan memublikasikannya ke topik Pub/Sub menggunakan Dataflow.

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

  • Hanya sel yang diubah dan deskripsi operasi penghapusan yang ditampilkan.
  • Hanya nilai baru dari sel yang diubah yang ditampilkan.

Saat data perubahan data dipublikasikan ke topik Pub/Sub, pesan mungkin disisipkan secara tidak berurutan dibandingkan dengan pengurutan stempel waktu commit Bigtable asli.

Data Bigtable yang tidak dapat dipublikasikan ke topik Pub/Sub sementara ditempatkan di direktori antrean pesan yang dihentikan pengirimannya (antrean pesan yang tidak diproses) di Cloud Storage. Setelah jumlah maksimum percobaan ulang yang gagal, kumpulan data ini akan ditempatkan tanpa batas waktu di direktori antrean surat tidak terkirim yang sama untuk ditinjau secara manual atau diproses lebih lanjut oleh pengguna.

Pipeline mengharuskan topik Pub/Sub tujuan ada. Topik tujuan dapat dikonfigurasi untuk memvalidasi pesan menggunakan skema. Saat topik Pub/Sub menentukan skema, pipeline hanya dimulai jika skema tersebut valid. Bergantung pada jenis skema, gunakan salah satu definisi skema berikut untuk topik tujuan:

Buffering protokol

syntax = "proto2";

package com.google.cloud.teleport.bigtable;

option java_outer_classname = "ChangeLogEntryProto";

message ChangelogEntryProto{
  required bytes rowKey = 1;
  enum ModType {
    SET_CELL = 0;
    DELETE_FAMILY = 1;
    DELETE_CELLS = 2;
    UNKNOWN = 3;
  }
  required ModType modType = 2;
  required bool isGC = 3;
  required int32 tieBreaker = 4;
  required int64 commitTimestamp = 5;
  required string columnFamily = 6;
  optional bytes column = 7;
  optional int64 timestamp = 8;
  optional int64 timestampFrom = 9;
  optional int64 timestampTo = 10;
  optional bytes value = 11;
  required string sourceInstance = 12;
  required string sourceCluster = 13;
  required string sourceTable = 14;
}

Avro

{
    "name" : "ChangelogEntryMessage",
    "type" : "record",
    "namespace" : "com.google.cloud.teleport.bigtable",
    "fields" : [
      { "name" : "rowKey", "type" : "bytes"},
      {
        "name" : "modType",
        "type" : {
          "name": "ModType",
          "type": "enum",
          "symbols": ["SET_CELL", "DELETE_FAMILY", "DELETE_CELLS", "UNKNOWN"]}
      },
      { "name": "isGC", "type": "boolean" },
      { "name": "tieBreaker", "type": "int"},
      { "name": "columnFamily", "type": "string"},
      { "name": "commitTimestamp", "type" : "long"},
      { "name" : "sourceInstance", "type" : "string"},
      { "name" : "sourceCluster", "type" : "string"},
      { "name" : "sourceTable", "type" : "string"},
      { "name": "column", "type" : ["null", "bytes"]},
      { "name": "timestamp", "type" : ["null", "long"]},
      { "name": "timestampFrom", "type" : ["null", "long"]},
      { "name": "timestampTo", "type" : ["null", "long"]},
      { "name" : "value", "type" : ["null", "bytes"]}
   ]
}

JSON

Gunakan skema Protobuf berikut dengan encoding pesan JSON:

syntax = "proto2";

package com.google.cloud.teleport.bigtable;

option java_outer_classname = "ChangelogEntryMessageText";

message ChangelogEntryText{
  required string rowKey = 1;
  enum ModType {
    SET_CELL = 0;
    DELETE_FAMILY = 1;
    DELETE_CELLS = 2;
    UNKNOWN = 3;
  }
  required ModType modType = 2;
  required bool isGC = 3;
  required int32 tieBreaker = 4;
  required int64 commitTimestamp = 5;
  required string columnFamily = 6;
  optional string column = 7;
  optional int64 timestamp = 8;
  optional int64 timestampFrom = 9;
  optional int64 timestampTo = 10;
  optional string value = 11;
  required string sourceInstance = 12;
  required string sourceCluster = 13;
  required string sourceTable = 14;
}

Setiap pesan Pub/Sub baru menyertakan satu entri dari data change record yang ditampilkan oleh aliran perubahan dari baris yang sesuai di tabel Bigtable Anda. Template Pub/Sub meratakan entri di setiap catatan perubahan data menjadi perubahan tingkat sel individual.

Deskripsi pesan output Pub/Sub

Nama kolom Deskripsi
rowKey Kunci baris dari baris yang diubah. Data ini diterima dalam bentuk array byte. Saat encoding pesan JSON dikonfigurasi, kunci baris ditampilkan sebagai string. Jika useBase64Rowkeys ditentukan, kunci baris akan dienkode Base64. Jika tidak, set karakter yang ditentukan oleh bigtableChangeStreamCharset akan digunakan untuk mendekode byte kunci baris menjadi string.
modType Jenis mutasi baris. Gunakan salah satu nilai berikut: SET_CELL, DELETE_CELLS, atau DELETE_FAMILY.
columnFamily Grup kolom yang terpengaruh oleh mutasi baris.
column Penentu kolom yang terpengaruh oleh mutasi baris. Untuk jenis mutasi DELETE_FAMILY, kolom kolom tidak ditetapkan. Data ini diterima dalam bentuk array byte. Saat encoding pesan JSON dikonfigurasi, kolom akan ditampilkan sebagai string. Jika useBase64ColumnQualifier ditentukan, kolom kolom akan dienkode Base64. Jika tidak, set karakter yang ditentukan oleh bigtableChangeStreamCharset akan digunakan untuk mendekode byte kunci baris menjadi string.
commitTimestamp Waktu saat Bigtable menerapkan mutasi. Waktu diukur dalam mikrodetik sejak epoch Unix (1 Januari 1970 pada UTC).
timestamp Nilai stempel waktu sel yang terpengaruh oleh mutasi. Untuk jenis mutasi DELETE_CELLS dan DELETE_FAMILY, stempel waktu tidak ditetapkan. Waktu diukur dalam mikrodetik sejak epoch Unix (1 Januari 1970 pada UTC).
timestampFrom Menjelaskan awal inklusif interval stempel waktu untuk semua sel yang dihapus oleh mutasi DELETE_CELLS. Untuk jenis mutasi lainnya, timestampFrom tidak ditetapkan. Waktu diukur dalam mikrodetik sejak epoch Unix (1 Januari 1970 pada UTC).
timestampTo Menjelaskan akhir eksklusif interval stempel waktu untuk semua sel yang dihapus oleh mutasi DELETE_CELLS. Untuk jenis mutasi lainnya, timestampTo tidak ditetapkan.
isGC Nilai boolean yang menunjukkan apakah mutasi dihasilkan oleh mekanisme pembersihan sampah Bigtable.
tieBreaker Jika dua mutasi didaftarkan secara bersamaan oleh cluster Bigtable yang berbeda, mutasi dengan nilai tiebreaker tertinggi akan diterapkan ke tabel sumber. Mutasi dengan nilai tiebreaker yang lebih rendah akan dihapus.
value Nilai baru yang ditetapkan oleh mutasi. Kecuali jika opsi pipeline stripValues ditetapkan, nilai akan ditetapkan untuk mutasi SET_CELL. Untuk jenis mutasi lainnya, nilai tidak ditetapkan. Data ini diterima dalam bentuk array byte. Saat encoding pesan JSON dikonfigurasi, nilai akan ditampilkan sebagai string. Jika useBase64Values ditentukan, nilainya akan dienkode Base64. Jika tidak, set karakter yang ditentukan oleh bigtableChangeStreamCharset akan digunakan untuk mendekode byte nilai menjadi string.
sourceInstance Nama instance Bigtable yang mendaftarkan mutasi. Mungkin terjadi saat beberapa pipeline melakukan streaming perubahan dari instance yang berbeda ke topik Pub/Sub yang sama.
sourceCluster Nama cluster Bigtable yang mendaftarkan mutasi. Dapat digunakan saat beberapa pipeline melakukan streaming perubahan dari instance yang berbeda ke topik Pub/Sub yang sama.
sourceTable Nama tabel Bigtable yang menerima mutasi. Dapat digunakan jika beberapa pipeline melakukan streaming perubahan dari tabel yang berbeda ke topik Pub/Sub yang sama.

Persyaratan pipeline

  • Instance sumber Bigtable yang ditentukan.
  • Tabel sumber Bigtable yang ditentukan. Tabel harus mengaktifkan aliran perubahan.
  • Profil aplikasi Bigtable yang ditentukan.
  • Topik Pub/Sub yang ditentukan harus ada.

Parameter template

Parameter yang diperlukan

  • pubSubTopic: Nama topik Pub/Sub tujuan.
  • bigtableChangeStreamAppProfile: ID profil aplikasi Bigtable. Profil aplikasi harus menggunakan perutean cluster tunggal dan mengizinkan transaksi baris tunggal.
  • bigtableReadInstanceId: ID instance Bigtable sumber.
  • bigtableReadTableId: ID tabel Bigtable sumber.

Parameter opsional

  • messageEncoding: Encoding pesan yang akan dipublikasikan ke topik Pub/Sub. Saat skema topik tujuan dikonfigurasi, encoding pesan ditentukan oleh setelan topik. Nilai berikut didukung: BINARY dan JSON. Setelan defaultnya adalah JSON.
  • messageFormat: Encoding pesan yang akan dipublikasikan ke topik Pub/Sub. Saat skema topik tujuan dikonfigurasi, encoding pesan ditentukan oleh setelan topik. Nilai berikut didukung: AVRO, PROTOCOL_BUFFERS, dan JSON. Nilai defaultnya adalah JSON. Saat format JSON digunakan, kolom rowKey, column, dan value dari pesan adalah string, yang kontennya ditentukan oleh opsi pipeline useBase64Rowkeys, useBase64ColumnQualifiers, useBase64Values, dan bigtableChangeStreamCharset.
  • stripValues: Jika ditetapkan ke true, mutasi SET_CELL akan ditampilkan tanpa menetapkan nilai baru. Setelan defaultnya adalah false. Parameter ini berguna jika Anda tidak memerlukan nilai baru, yang juga dikenal sebagai pembatalan validasi cache, atau jika nilai sangat besar dan melebihi batas ukuran pesan Pub/Sub.
  • dlqDirectory: Direktori untuk antrean surat mati. Data yang gagal diproses disimpan di direktori ini. Setelan default-nya adalah direktori di lokasi sementara tugas Dataflow. Dalam sebagian besar kasus, Anda dapat menggunakan jalur default.
  • dlqRetryMinutes: Jumlah menit antara percobaan ulang antrean pesan yang tidak terkirim. Default-nya adalah 10.
  • dlqMaxRetries: Percobaan ulang maksimum untuk pesan yang pengirimannya dihentikan. Default-nya adalah 5.
  • useBase64Rowkeys: Digunakan dengan encoding pesan JSON. Jika disetel ke true, kolom rowKey adalah string berenkode Base64. Jika tidak, rowKey akan dihasilkan dengan menggunakan bigtableChangeStreamCharset untuk mendekode byte menjadi string. Nilai default-nya adalah false.
  • pubSubProjectId: Project ID Bigtable. Defaultnya adalah project tugas Dataflow.
  • useBase64ColumnQualifiers: Digunakan dengan encoding pesan JSON. Jika disetel ke true, kolom column adalah string berenkode Base64. Jika tidak, kolom akan dibuat menggunakan bigtableChangeStreamCharset untuk mendekode byte menjadi string. Setelan defaultnya adalah false.
  • useBase64Values: Digunakan dengan encoding pesan JSON. Jika ditetapkan ke true, kolom nilai adalah string yang dienkode Base64. Jika tidak, nilai akan dihasilkan dengan menggunakan bigtableChangeStreamCharset untuk mendekode byte menjadi string. Setelan defaultnya adalah false.
  • disableDlqRetries: Apakah akan menonaktifkan percobaan ulang untuk DLQ atau tidak. Defaultnya adalah: false.
  • bigtableChangeStreamMetadataInstanceId: ID instance metadata aliran perubahan data Bigtable. Default-nya adalah kosong.
  • bigtableChangeStreamMetadataTableTableId: ID tabel metadata konektor aliran perubahan data Bigtable. Jika tidak diberikan, tabel metadata konektor aliran perubahan Bigtable akan otomatis dibuat selama eksekusi pipeline. Default-nya adalah kosong.
  • bigtableChangeStreamCharset: Nama set karakter aliran perubahan data Bigtable. Setelan defaultnya adalah: UTF-8.
  • bigtableChangeStreamStartTimestamp: Stempel waktu awal (https://tools.ietf.org/html/rfc3339), inklusif, untuk digunakan membaca aliran perubahan. Misalnya, 2022-05-05T07:59:59Z. Default-nya adalah stempel waktu waktu mulai pipeline.
  • bigtableChangeStreamIgnoreColumnFamilies: Daftar perubahan nama grup kolom yang dipisahkan koma yang akan diabaikan. Default-nya adalah kosong.
  • bigtableChangeStreamIgnoreColumns: Daftar perubahan nama kolom yang dipisahkan koma yang akan diabaikan. Default-nya adalah kosong.
  • bigtableChangeStreamName: Nama unik untuk pipeline klien. Memungkinkan Anda melanjutkan pemrosesan dari titik saat pipeline yang sebelumnya berjalan dihentikan. Secara default, nama yang dibuat secara otomatis. Lihat log tugas Dataflow untuk mengetahui nilai yang digunakan.
  • bigtableChangeStreamResume: 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. Default-nya adalah false.
  • bigtableReadProjectId: Project ID Bigtable. Defaultnya adalah project untuk tugas Dataflow.

Menjalankan template

Konsol

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

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

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

gcloud

Di shell atau terminal, 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_PubSub \
    --parameters \
bigtableReadInstanceId=BIGTABLE_INSTANCE_ID,\
bigtableReadTableId=BIGTABLE_TABLE_ID,\
bigtableChangeStreamAppProfile=BIGTABLE_APPLICATION_PROFILE_ID,\
pubSubTopic=PUBSUB_TOPIC

Ganti kode berikut:

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

    Anda dapat menggunakan nilai berikut:

  • 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.
  • PUBSUB_TOPIC: nama topik tujuan Pub/Sub

API

Untuk menjalankan template menggunakan REST API, kirim permintaan POST HTTP. Untuk mengetahui 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_PubSub",
    "parameters": {
        "bigtableReadInstanceId": "BIGTABLE_INSTANCE_ID",
        "bigtableReadTableId": "BIGTABLE_TABLE_ID",
        "bigtableChangeStreamAppProfile": "BIGTABLE_APPLICATION_PROFILE_ID",
        "pubSubTopic": "PUBSUB_TOPIC"
    }
  }
}

Ganti kode berikut:

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

    Anda dapat menggunakan nilai berikut:

  • 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.
  • PUBSUB_TOPIC: nama topik tujuan Pub/Sub

Langkah berikutnya