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

Mengalirkan perubahan dengan Dataflow

Konektor Bigtable Beam memungkinkan Anda menggunakan Dataflow untuk membaca data Bigtable yang mengubah catatan tanpa perlu melacak atau memproses perubahan partisi dalam kode Anda, karena konektor menangani logika tersebut untuk Anda.

Dokumen ini menjelaskan cara mengonfigurasi dan menggunakan konektor Bigtable Beam untuk membaca aliran perubahan menggunakan pipeline Dataflow. Sebelum membaca dokumen ini, Anda harus membaca Ringkasan aliran perubahan dan memahami Dataflow.

Alternatif untuk membuat pipeline Anda sendiri

Jika tidak ingin membuat pipeline Dataflow Anda sendiri, Anda dapat menggunakan salah satu opsi berikut.

Anda dapat menggunakan template Dataflow yang disediakan Google.

Anda juga dapat menggunakan contoh kode dari tutorial atau panduan memulai Bigtable sebagai titik awal untuk kode Anda.

Pastikan kode yang Anda buat menggunakan google cloud libraries-bom versi 26.14.0 atau yang lebih baru.

Detail konektor

Metode konektor Bigtable Beam, BigtableIO.readChangeStream, memungkinkan Anda membaca aliran catatan perubahan data (ChangeStreamMutation) yang dapat Anda proses. Konektor Bigtable Beam adalah komponen repositori GitHub Apache Beam. Untuk mengetahui deskripsi kode konektor, lihat komentar di BigtableIO.java.

Anda harus menggunakan konektor dengan Beam versi 2.48.0 atau yang lebih baru. Periksa Apache Beam dukungan runtime untuk memastikan bahwa Anda menggunakan Java versi yang didukung. Kemudian, Anda dapat men-deploy pipeline yang menggunakan konektor ke Dataflow, yang menangani penyediaan dan pengelolaan resource serta membantu skalabilitas dan keandalan pemrosesan data streaming.

Untuk mengetahui informasi selengkapnya tentang model pemrograman Apache Beam, lihat dokumentasi Beam.

Mengelompokkan data tanpa waktu peristiwa

Catatan perubahan data yang di-streaming menggunakan konektor Bigtable Beam tidak kompatibel dengan fungsi Dataflow yang bergantung pada waktu peristiwa.

Seperti yang dijelaskan dalam Replikasi dan watermark, watermark rendah mungkin tidak akan bertambah jika replikasi untuk partisi belum menyusul bagian instance lainnya. Jika watermark rendah berhenti bertambah, hal ini dapat menyebabkan aliran perubahan terhenti.

Untuk mencegah aliran terhenti, konektor Bigtable Beam menampilkan semua data dengan stempel waktu output nol. Stempel waktu nol membuat Dataflow menganggap semua catatan perubahan data sebagai data terlambat. Akibatnya, fitur Dataflow yang bergantung pada waktu peristiwa tidak kompatibel dengan aliran perubahan Bigtable. Secara khusus, Anda tidak dapat menggunakan fungsi jendela, pemicu waktu peristiwa, atau timer waktu peristiwa.

Sebagai gantinya, Anda dapat menggunakan GlobalWindows dengan pemicu waktu non-peristiwa untuk mengelompokkan data terlambat ini ke dalam panel, seperti yang ditunjukkan dalam contoh dari tutorial. Untuk mengetahui detail tentang pemicu dan panel, lihat Pemicu dalam panduan pemrograman Beam.

Penskalaan otomatis

Konektor mendukung pensakalaan otomatis Dataflow, yang diaktifkan secara default saat menggunakan Portable Runner (wajib). Algoritma penskalaan otomatis Dataflow mempertimbangkan backlog aliran perubahan yang diperkirakan, yang dapat dipantau di halaman pemantauan Dataflow di bagian Backlog. Gunakan flag --maxNumWorkers saat men-deploy tugas untuk membatasi jumlah pekerja.

Untuk menskalakan pipeline secara manual, bukan menggunakan penskalaan otomatis, lihat Menskalakan pipeline streaming secara manual.

Batasan

Perhatikan batasan berikut sebelum menggunakan konektor Bigtable Beam dengan Dataflow.

Dataflow Portable Runner

Konektor hanya dapat dieksekusi menggunakan Dataflow Portable Runner. Untuk mengaktifkan hal ini, tentukan --experiments=use_runner_v2 dalam argumen command line Anda. Menjalankan dengan Streaming Java Runner akan menyebabkan pipeline Anda gagal dengan pengecualian berikut:

java.lang.UnsupportedOperationException: BundleFinalizer unsupported by non-portable Dataflow

Snapshot

Konektor tidak mendukung snapshot Dataflow.

Duplikat

Konektor Bigtable Beam mengalirkan perubahan untuk setiap row key dan setiap cluster dalam urutan stempel waktu commit, tetapi karena terkadang dimulai ulang dari waktu sebelumnya dalam aliran, konektor dapat menghasilkan duplikat.

Pipeline dimulai ulang

Jika pipeline Dataflow telah berhenti dalam waktu yang lama, catatan perubahan data dapat tertinggal dari batas retensi. Saat pipeline dilanjutkan, Bigtable akan menggagalkan pipeline sehingga Anda dapat memulai pipeline baru dengan waktu mulai permintaan baru yang berada dalam periode retensi data. Bigtable melakukan hal ini, bukan secara diam-diam memajukan waktu permintaan pipeline asli, untuk mencegah penghapusan catatan perubahan data yang tidak disengaja dengan stempel waktu yang berada di luar periode retensi data yang ditentukan.

Sebelum memulai

Sebelum menggunakan konektor, selesaikan prasyarat berikut.

Menyiapkan autentikasi

Untuk menggunakan contoh Java di halaman ini dalam lingkungan pengembangan lokal, instal dan lakukan inisialisasi gcloud CLI, lalu siapkan Kredensial Default Aplikasi dengan kredensial pengguna Anda.

Instal Google Cloud CLI.
Konfigurasi gcloud CLI agar menggunakan identitas gabungan Anda.

Untuk mengetahui informasi selengkapnya, lihat Login ke gcloud CLI dengan identitas gabungan Anda.
Buat kredensial autentikasi lokal untuk akun pengguna Anda:
```
gcloud auth application-default login
```
Jika error autentikasi ditampilkan, dan Anda menggunakan penyedia identitas (IdP) eksternal, pastikan Anda telah login ke gcloud CLI dengan identitas gabungan Anda.

Untuk mengetahui informasi selengkapnya, lihat Menyiapkan autentikasi untuk lingkungan pengembangan lokal.

Untuk mengetahui informasi tentang cara menyiapkan autentikasi untuk lingkungan produksi, lihat Menyiapkan Kredensial Default Aplikasi untuk kode yang berjalan di Google Cloud .

Mengaktifkan aliran perubahan

Anda harus mengaktifkan aliran perubahan pada tabel sebelum dapat membacanya. Anda juga dapat membuat tabel baru dengan aliran perubahan yang diaktifkan.

Tabel metadata aliran perubahan

Saat Anda melakukan streaming perubahan dengan Dataflow, konektor Bigtable Beam akan membuat tabel metadata yang diberi nama __change_stream_md_table secara default. Tabel metadata aliran perubahan mengelola status operasional konektor dan menyimpan metadata tentang catatan perubahan data.

Secara default, konektor membuat tabel di instance yang sama dengan tabel yang di-streaming. Untuk memastikan tabel berfungsi dengan benar, profil aplikasi untuk tabel metadata harus menggunakan perutean cluster tunggal dan mengaktifkan transaksi baris tunggal.

Untuk mengetahui informasi selengkapnya tentang streaming perubahan dari Bigtable dengan konektor Bigtable Beam, lihat dokumentasi BigtableIO.

Peran yang diperlukan

Untuk mendapatkan izin yang Anda perlukan untuk membaca aliran perubahan Bigtable menggunakan Dataflow, minta administrator untuk memberi Anda peran IAM berikut.

Untuk membaca perubahan dari Bigtable, Anda memerlukan peran ini:

Bigtable Administrator (roles/bigtable.admin) di instance Bigtable yang berisi tabel yang akan Anda gunakan untuk melakukan streaming perubahan

Untuk menjalankan tugas Dataflow, Anda memerlukan peran ini:

Dataflow Developer (roles/dataflow.developer) di project yang berisi resource Cloud Anda
Dataflow Worker (roles/dataflow.worker) di project yang berisi resource Cloud Anda
Storage Object Admin (roles/storage.objectAdmin) di bucket Cloud Storage yang akan Anda gunakan

Untuk mengetahui informasi selengkapnya tentang cara memberikan peran, lihat Mengelola akses.

Anda mungkin juga bisa mendapatkan izin yang diperlukan melalui peran khusus atau peran bawaan lainnya.

Menambahkan konektor Bigtable Beam sebagai dependensi

Tambahkan kode yang mirip dengan dependensi berikut ke file pom.xml Maven Anda. Versinya harus 2.48.0 atau yang lebih baru.

<dependencies>
  <dependency>
    <groupId>org.apache.beam</groupId>
    <artifactId>beam-sdks-java-io-google-cloud-platform</artifactId>
    <version>VERSION</version>
  </dependency>
</dependencies>

Membaca aliran perubahan

Untuk membuat pipeline Dataflow guna membaca catatan perubahan data Anda, konfigurasikan konektor, lalu tambahkan transformasi dan sink. Kemudian, gunakan konektor untuk membaca objek ChangeStreamMutation dalam pipeline Beam.

Contoh kode di bagian ini, yang ditulis dalam Java, menunjukkan cara membuat pipeline dan menggunakannya untuk mengonversi pasangan nilai kunci menjadi string. Setiap pasangan terdiri dari row key dan objek ChangeStreamMutation. Pipeline mengonversi setiap entri objek menjadi string yang dipisahkan koma.

Membangun pipeline

Contoh kode Java ini menunjukkan cara membangun pipeline:

BigtableOptions options =
    PipelineOptionsFactory.fromArgs(args).withValidation().as(BigtableOptions.class);
Pipeline p = Pipeline.create(options);

final Instant startTime = Instant.now();

p.apply(
        "Read Change Stream",
        BigtableIO.readChangeStream()
            .withProjectId(options.getBigtableProjectId())
            .withInstanceId(options.getBigtableInstanceId())
            .withTableId(options.getBigtableTableId())
            .withAppProfileId(options.getBigtableAppProfile())
            .withStartTime(startTime))
    .apply(
        "Flatten Mutation Entries",
        FlatMapElements.into(TypeDescriptors.strings())
            .via(ChangeStreamsHelloWorld::mutationEntriesToString))
    .apply(
        "Print mutations",
        ParDo.of(
            new DoFn<String, Void>() { // a DoFn as an anonymous inner class instance
              @ProcessElement
              public void processElement(@Element String mutation) {
                System.out.println("Change captured: " + mutation);
              }
            }));
p.run();

Memproses catatan perubahan data

Contoh ini menunjukkan cara melakukan loop melalui semua entri dalam catatan perubahan data untuk baris dan memanggil metode konversi ke string berdasarkan jenis entri.

Untuk mengetahui daftar jenis entri yang dapat berisi catatan perubahan data, lihat Isi catatan perubahan data.

static List<String> mutationEntriesToString(KV<ByteString, ChangeStreamMutation> mutationPair) {
  List<String> mutations = new ArrayList<>();
  String rowKey = mutationPair.getKey().toStringUtf8();
  ChangeStreamMutation mutation = mutationPair.getValue();
  MutationType mutationType = mutation.getType();
  for (Entry entry : mutation.getEntries()) {
    if (entry instanceof SetCell) {
      mutations.add(setCellToString(rowKey, mutationType, (SetCell) entry));
    } else if (entry instanceof DeleteCells) {
      mutations.add(deleteCellsToString(rowKey, mutationType, (DeleteCells) entry));
    } else if (entry instanceof DeleteFamily) {
      // Note: DeleteRow mutations are mapped into one DeleteFamily per-family
      mutations.add(deleteFamilyToString(rowKey, mutationType, (DeleteFamily) entry));
    } else {
      throw new RuntimeException("Entry type not supported.");
    }
  }
  return mutations;
}

Dalam contoh ini, entri tulis dikonversi:

private static String setCellToString(String rowKey, MutationType mutationType, SetCell setCell) {
  List<String> mutationParts =
      Arrays.asList(
          rowKey,
          mutationType.name(),
          "SetCell",
          setCell.getFamilyName(),
          setCell.getQualifier().toStringUtf8(),
          setCell.getValue().toStringUtf8());
  return String.join(",", mutationParts);
}

Dalam contoh ini, entri penghapusan sel dikonversi:

private static String deleteCellsToString(
    String rowKey, MutationType mutationType, DeleteCells deleteCells) {
  String timestampRange =
      deleteCells.getTimestampRange().getStart() + "-" + deleteCells.getTimestampRange().getEnd();
  List<String> mutationParts =
      Arrays.asList(
          rowKey,
          mutationType.name(),
          "DeleteCells",
          deleteCells.getFamilyName(),
          deleteCells.getQualifier().toStringUtf8(),
          timestampRange);
  return String.join(",", mutationParts);
}

Dalam contoh ini, entri penghapusan grup kolom dikonversi:


private static String deleteFamilyToString(
    String rowKey, MutationType mutationType, DeleteFamily deleteFamily) {
  List<String> mutationParts =
      Arrays.asList(rowKey, mutationType.name(), "DeleteFamily", deleteFamily.getFamilyName());
  return String.join(",", mutationParts);
}

Memantau

Resource berikut di Google Cloud konsol memungkinkan Anda memantau Google Cloud resource saat menjalankan pipeline Dataflow untuk membaca aliran perubahan Bigtable:

Secara khusus, periksa metrik berikut:

Di halaman insight sistem Bigtable, periksa metrik berikut :
- Data penggunaan CPU menurut aliran perubahan dalam metrik cpu_load_by_app_profile_by_method_by_table. Menunjukkan dampak aliran perubahan pada penggunaan CPU cluster Anda.
- Penggunaan penyimpanan aliran perubahan (byte) (change_stream_log_used_bytes).
Di halaman pemantauan Dataflow, periksa keaktualan data. Metrik ini menunjukkan perbedaan antara waktu saat ini dan watermark, yang kira-kira dua menit, dengan lonjakan sesekali yang satu atau dua menit lebih lama. Keaktualan data tidak menunjukkan apakah catatan perubahan data diproses secara lambat. Untuk memastikan kesehatan dan performa aplikasi penting Anda yang berkelanjutan, pantau metrik keaktualan data Dataflow dan lakukan tindakan berikut:
- Jika metrik keaktualan data secara konsisten lebih tinggi dari nilai minimum, pipeline Anda mungkin kekurangan resource. Sebaiknya tambahkan lebih banyak pekerja Dataflow.
- Jika pekerja Dataflow disediakan dengan baik, tetapi keaktualan data terus meningkat atau secara konsisten tinggi, hubungi Google Cloud Dukungan.
Metrik Dataflow processing_delay_from_commit_timestamp_MEAN dapat memberi tahu Anda waktu pemrosesan rata-rata catatan perubahan data selama masa aktif tugas.

Metrik Bigtable server/latencies tidak berguna saat Anda memantau pipeline Dataflow yang membaca aliran perubahan Bigtable, karena metrik ini mencerminkan durasi permintaan streaming, bukan latensi pemrosesan catatan perubahan data. Latensi tinggi dalam aliran perubahan tidak berarti permintaan diproses secara lambat; artinya koneksi terbuka selama itu.