Memproses aliran data perubahan Bigtable

---

Tutorial ini menunjukkan cara men-deploy pipeline data ke Dataflow untuk aliran perubahan database real-time yang bersumber dari aliran perubahan tabel Bigtable. Output pipeline ditulis ke serangkaian file di Cloud Storage.

Contoh set data untuk aplikasi pemutaran musik disediakan. Dalam tutorial ini, Anda akan melacak lagu yang didengarkan, lalu menentukan peringkat lima teratas selama periode tertentu.

Tutorial ini ditujukan untuk pengguna teknis yang sudah memahami cara menulis kode dan men-deploy pipeline data ke Google Cloud.

Tujuan

Tutorial ini menunjukkan cara melakukan hal berikut:

• Buat tabel Bigtable dengan aliran data perubahan diaktifkan.
• Men-deploy pipeline di Dataflow yang mengubah dan menghasilkan aliran perubahan.
• Lihat hasil pipeline data Anda.

Biaya

Dalam dokumen ini, Anda akan menggunakan komponen Google Cloud yang dapat ditagih berikut:

• Bigtable
• Compute Engine
• Dataflow
• Cloud Storage

Untuk membuat perkiraan biaya berdasarkan proyeksi penggunaan Anda, gunakan kalkulator harga. Pengguna baru Google Cloud mungkin memenuhi syarat untuk mendapatkan uji coba gratis.

Setelah menyelesaikan tugas yang dijelaskan dalam dokumen ini, Anda dapat menghindari penagihan berkelanjutan dengan menghapus resource yang Anda buat. Untuk mengetahui informasi selengkapnya, lihat Pembersihan.

Sebelum memulai

1. Sign in to your Google Cloud account. If you're new to Google Cloud, create an account to evaluate how our products perform in real-world scenarios. New customers also get $300 in free credits to run, test, and deploy workloads.
2. Install the Google Cloud CLI.

3. To initialize the gcloud CLI, run the following command:

   gcloud init

4. Create or select a Google Cloud project.

   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   • Create a Google Cloud project:

     gcloud projects create PROJECT_ID

     Replace PROJECT_ID with a name for the Google Cloud project you are creating.

   • Select the Google Cloud project that you created:

     gcloud config set project PROJECT_ID

     Replace PROJECT_ID with your Google Cloud project name.

5. Make sure that billing is enabled for your Google Cloud project.

6. Enable the Dataflow, Cloud Bigtable API, Cloud Bigtable Admin API, and Cloud Storage APIs:

   gcloud services enable dataflow.googleapis.com bigtable.googleapis.com bigtableadmin.googleapis.com storage.googleapis.com

7. Install the Google Cloud CLI.

8. To initialize the gcloud CLI, run the following command:

   gcloud init

9. Create or select a Google Cloud project.

   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   • Create a Google Cloud project:

     gcloud projects create PROJECT_ID

     Replace PROJECT_ID with a name for the Google Cloud project you are creating.

   • Select the Google Cloud project that you created:

     gcloud config set project PROJECT_ID

     Replace PROJECT_ID with your Google Cloud project name.

10. Make sure that billing is enabled for your Google Cloud project.

11. Enable the Dataflow, Cloud Bigtable API, Cloud Bigtable Admin API, and Cloud Storage APIs:

    gcloud services enable dataflow.googleapis.com bigtable.googleapis.com bigtableadmin.googleapis.com storage.googleapis.com

12. Update dan instal cbt CLI.

    gcloud components update
    gcloud components install cbt

Menyiapkan lingkungan

Mendapatkan kode

Buat clone repositori yang berisi kode contoh. Jika sebelumnya Anda telah mendownload repositori ini, lakukan pull untuk mendapatkan versi terbaru.

git clone https://github.com/GoogleCloudPlatform/java-docs-samples.git
cd java-docs-samples/bigtable/beam/change-streams

Membuat bucket

Create a Cloud Storage bucket:

gcloud storage buckets create gs://BUCKET_NAME

Replace BUCKET_NAME with a bucket name that meets the bucket naming requirements.

Membuat instance Bigtable

Anda dapat menggunakan instance yang ada untuk tutorial ini atau membuat instance dengan konfigurasi default di region yang dekat dengan Anda.

Membuat tabel

Aplikasi contoh melacak lagu yang didengarkan pengguna dan menyimpan peristiwa dengar di Bigtable. Buat tabel dengan aliran perubahan yang diaktifkan yang memiliki satu keluarga kolom (cf) dan satu kolom (song) dan menggunakan ID pengguna untuk kunci baris.

Buat tabel.

gcloud bigtable instances tables create song-rank \
--column-families=cf --change-stream-retention-period=7d \
--instance=BIGTABLE_INSTANCE_ID --project=PROJECT_ID

Ganti kode berikut:

• PROJECT_ID: ID project yang Anda gunakan
• BIGTABLE_INSTANCE_ID: ID instance yang akan berisi tabel baru

Memulai pipeline

Pipeline ini mengubah aliran perubahan dengan melakukan hal berikut:

1. Membaca aliran data perubahan
2. Mendapatkan nama lagu
3. Mengelompokkan peristiwa pemutaran lagu ke dalam periode N detik
4. Menghitung lima lagu teratas
5. Menampilkan hasil

Jalankan pipeline.

mvn compile exec:java -Dexec.mainClass=SongRank \
"-Dexec.args=--project=PROJECT_ID --bigtableProjectId=PROJECT_ID \
--bigtableInstanceId=BIGTABLE_INSTANCE_ID --bigtableTableId=song-rank \
--outputLocation=gs://BUCKET_NAME/ \
--runner=dataflow --region=BIGTABLE_REGION --experiments=use_runner_v2"

Ganti BIGTABLE_REGION dengan ID region tempat instance Bigtable Anda berada, seperti us-east5.

Memahami pipeline

Cuplikan kode berikut dari pipeline dapat membantu Anda memahami kode yang Anda jalankan.

Membaca aliran data perubahan

Kode dalam contoh ini mengonfigurasi aliran sumber dengan parameter untuk instance dan tabel Bigtable tertentu.

p.apply(
        "Stream from Bigtable",
        BigtableIO.readChangeStream()
            .withProjectId(options.getBigtableProjectId())
            .withInstanceId(options.getBigtableInstanceId())
            .withTableId(options.getBigtableTableId())
            .withAppProfileId(options.getBigtableAppProfile())

    )

Mendapatkan nama lagu

Saat lagu didengarkan, nama lagu akan ditulis ke grup kolom cf dan penentu kolom song, sehingga kode mengekstrak nilai dari mutasi streaming perubahan dan menampilkannya ke langkah berikutnya dari pipeline.

private static class ExtractSongName extends DoFn<KV<ByteString, ChangeStreamMutation>, String> {

  @DoFn.ProcessElement
  public void processElement(ProcessContext c) {

    for (Entry e : Objects.requireNonNull(Objects.requireNonNull(c.element()).getValue())
        .getEntries()) {
      if (e instanceof SetCell) {
        SetCell setCell = (SetCell) e;
        if ("cf".equals(setCell.getFamilyName())
            && "song".equals(setCell.getQualifier().toStringUtf8())) {
          c.output(setCell.getValue().toStringUtf8());
        }
      }
    }
  }
}

Menghitung lima lagu teratas

Anda dapat menggunakan fungsi Beam bawaan Count dan Top.of untuk mendapatkan lima lagu teratas di jendela saat ini.

.apply(Count.perElement())
.apply("Top songs", Top.of(5, new SongComparator()).withoutDefaults())

Menampilkan hasil

Pipeline ini menulis hasil ke output standar serta file. Untuk file, penulisan akan dikelompokkan ke dalam grup yang terdiri dari 10 elemen atau segmen satu menit.

.apply("Print", ParDo.of(new PrintFn()))
.apply(
    "Collect at least 10 elements or 1 minute of elements",
    Window.<String>into(new GlobalWindows())
        .triggering(
            Repeatedly.forever(
                AfterFirst.of(
                    AfterPane.elementCountAtLeast(10),
                    AfterProcessingTime
                        .pastFirstElementInPane()
                        .plusDelayOf(Duration.standardMinutes(1)
                        )
                )
            ))
        .discardingFiredPanes())
.apply(
    "Output top songs",
    TextIO.write()
        .to(options.getOutputLocation() + "song-charts/")
        .withSuffix(".txt")
        .withNumShards(1)
        .withWindowedWrites()
);

Melihat pipeline

1. Di konsol Google Cloud, buka halaman Dataflow.

   Buka Dataflow

2. Klik tugas dengan nama yang dimulai dengan song-rank.

3. Di bagian bawah layar, klik Tampilkan untuk membuka panel log.

4. Klik Log pekerja untuk memantau log output aliran perubahan.

   Catatan: Log mungkin memerlukan waktu beberapa menit untuk muncul saat tugas diinisialisasi.

Operasi tulis streaming

Gunakan CLI cbt untuk menulis sejumlah lagu yang didengarkan oleh berbagai pengguna ke tabel song-rank. Ini dirancang untuk menulis selama beberapa menit untuk menyimulasikan streaming lagu yang didengarkan dari waktu ke waktu.

cbt -instance=BIGTABLE_INSTANCE_ID -project=PROJECT_ID import \
song-rank song-rank-data.csv  column-family=cf batch-size=1

Melihat output

Baca output di Cloud Storage untuk melihat lagu yang paling populer.

gcloud storage cat gs://BUCKET_NAME/song-charts/GlobalWindow-pane-0-00000-of-00001.txt

Catatan: File mungkin memerlukan waktu beberapa menit untuk muncul saat pipeline memproses data. Lihat log pekerja Dataflow untuk memantau statusnya.

Contoh output:

2023-07-06T19:53:38.232Z [KV{The Wheels on the Bus, 199}, KV{Twinkle, Twinkle, Little Star, 199}, KV{Ode to Joy , 192}, KV{Row, Row, Row Your Boat, 186}, KV{Take Me Out to the Ball Game, 182}]
2023-07-06T19:53:49.536Z [KV{Old MacDonald Had a Farm, 20}, KV{Take Me Out to the Ball Game, 18}, KV{Für Elise, 17}, KV{Ode to Joy , 15}, KV{Mary Had a Little Lamb, 12}]
2023-07-06T19:53:50.425Z [KV{Twinkle, Twinkle, Little Star, 20}, KV{The Wheels on the Bus, 17}, KV{Row, Row, Row Your Boat, 13}, KV{Happy Birthday to You, 12}, KV{Over the Rainbow, 9}]

Pembersihan

Agar tidak perlu membayar biaya pada akun Google Cloud Anda untuk resource yang digunakan dalam tutorial ini, hapus project yang berisi resource tersebut, atau simpan project dan hapus setiap resource.

Menghapus project

Perhatian: Menghapus project memiliki efek berikut:

• Semua hal dalam project akan dihapus. Jika menggunakan project yang sudah ada untuk tugas dalam dokumen ini, saat Anda menghapusnya, pekerjaan lain yang telah Anda lakukan dalam project tersebut juga akan terhapus.
• Project ID kustom hilang. Saat membuat project ini, Anda mungkin telah membuat project ID kustom yang ingin digunakan di masa mendatang. Untuk mempertahankan URL yang menggunakan project ID, seperti URL appspot.com, hapus resource yang dipilih di dalam project, bukan menghapus seluruh project.

Jika Anda berencana mempelajari beberapa arsitektur, tutorial atau panduan memulai, dengan menggunakan kembali project dapat membantu Anda agar tidak melampaui batas kuota project.

Delete a Google Cloud project:

gcloud projects delete PROJECT_ID

Menghapus resource satu per satu

1. Hapus bucket dan file.

   gcloud storage rm --recursive gs://BUCKET_NAME/
   

2. Nonaktifkan aliran perubahan pada tabel.

   gcloud bigtable instances tables update song-rank --instance=BIGTABLE_INSTANCE_ID \
   --clear-change-stream-retention-period
   

3. Hapus tabel song-rank.

   cbt -instance=BIGTABLE_INSTANCE_ID -project=PROJECT_ID deletetable song-rank
   

4. Hentikan pipeline aliran perubahan.

   1. Buat daftar tugas untuk mendapatkan ID tugas.

      gcloud dataflow jobs list --region=BIGTABLE_REGION
      

   2. Batalkan tugas.

      gcloud dataflow jobs cancel JOB_ID --region=BIGTABLE_REGION
      

      Ganti JOB_ID dengan ID tugas yang ditampilkan setelah perintah sebelumnya.

Langkah selanjutnya

• Baca dokumentasi aliran perubahan.
• Download kode contoh.