Memproses aliran data perubahan Bigtable


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

Contoh {i>dataset<i} untuk aplikasi mendengarkan musik telah disediakan. Dalam tutorial ini, Anda melacak lagu yang didengarkan, lalu memberi peringkat lima lagu teratas dalam satu periode.

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

Tujuan

Tutorial ini menunjukkan cara melakukan hal berikut:

  • Membuat tabel Bigtable dengan aliran data perubahan diaktifkan.
  • Deploy pipeline di Dataflow yang mentransformasi dan menghasilkan aliran perubahan.
  • Lihat hasil pipeline data Anda.

Biaya

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

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. Login ke akun Google Cloud Anda. Jika Anda baru menggunakan Google Cloud, buat akun untuk mengevaluasi performa produk kami dalam skenario dunia nyata. Pelanggan baru juga mendapatkan kredit gratis senilai $300 untuk menjalankan, menguji, dan men-deploy workload.
  2. Menginstal Google Cloud CLI.
  3. Untuk initialize gcloud CLI, jalankan perintah berikut:

    gcloud init
  4. Buat atau pilih project Google Cloud.

    • Membuat project Google Cloud:

      gcloud projects create PROJECT_ID

      Ganti PROJECT_ID dengan nama untuk project Google Cloud yang Anda buat.

    • Pilih project Google Cloud yang Anda buat:

      gcloud config set project PROJECT_ID

      Ganti PROJECT_ID dengan nama project Google Cloud Anda.

  5. Pastikan penagihan telah diaktifkan untuk project Google Cloud Anda.

  6. Aktifkan API Dataflow, Cloud Bigtable API, Cloud Bigtable Admin API, and Cloud Storage:

    gcloud services enable dataflow.googleapis.com bigtable.googleapis.com bigtableadmin.googleapis.com storage.googleapis.com
  7. Menginstal Google Cloud CLI.
  8. Untuk initialize gcloud CLI, jalankan perintah berikut:

    gcloud init
  9. Buat atau pilih project Google Cloud.

    • Membuat project Google Cloud:

      gcloud projects create PROJECT_ID

      Ganti PROJECT_ID dengan nama untuk project Google Cloud yang Anda buat.

    • Pilih project Google Cloud yang Anda buat:

      gcloud config set project PROJECT_ID

      Ganti PROJECT_ID dengan nama project Google Cloud Anda.

  10. Pastikan penagihan telah diaktifkan untuk project Google Cloud Anda.

  11. Aktifkan API Dataflow, Cloud Bigtable API, Cloud Bigtable Admin API, and Cloud Storage:

    gcloud services enable dataflow.googleapis.com bigtable.googleapis.com bigtableadmin.googleapis.com storage.googleapis.com
  12. Update dan instal CLI cbt.
    gcloud components update
    gcloud components install cbt
    

Menyiapkan lingkungan

Mendapatkan kode

Buat clone repositori yang berisi kode contoh. Jika sebelumnya Anda sudah mendownload repositori ini, tarik untuk mendapatkan versi terbaru.

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

Membuat bucket

  • Buat bucket Cloud Storage:
    gcloud storage buckets create gs://BUCKET_NAME
    Ganti BUCKET_NAME dengan nama bucket yang memenuhi persyaratan penamaan bucket.
  • Membuat instance Bigtable

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

    Membuat tabel

    Aplikasi contoh ini melacak lagu yang didengarkan pengguna dan menyimpan peristiwa pemrosesannya di Bigtable. Buat tabel dengan aliran perubahan aktif yang memiliki satu grup kolom (cf) dan satu kolom (lagu), serta menggunakan ID pengguna untuk tombol 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 untuk memuat tabel baru

    Memulai pipeline

    Pipeline ini mentransformasi aliran perubahan dengan melakukan hal berikut:

    1. Membaca aliran data perubahan
    2. Mendapatkan judul lagu
    3. Mengelompokkan peristiwa mendengarkan lagu ke dalam jendela 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 dari pipeline berikut dapat membantu Anda memahami kode yang Anda jalankan.

    Membaca aliran data perubahan

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

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

    Mendapatkan judul lagu

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

    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 Count dan Top.of bawaan 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 hasilnya ke standard out serta file. Untuk file, Windows akan jendela operasi tulis 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 diawali dengan peringkat lagu.

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

    4. Klik Worker logs untuk memantau log output dari aliran data perubahan.

    Penulisan streaming

    Gunakan CLI cbt untuk menulis sejumlah pemutaran lagu bagi berbagai pengguna ke tabel song-rank. Fungsi ini dirancang untuk menulis selama beberapa menit guna menyimulasikan streaming pemrosesan lagu 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 terpopuler.

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

    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

      Menghapus project Google Cloud:

      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 data 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. Tampilkan daftar tugas untuk mendapatkan ID tugas.

        gcloud dataflow jobs list --region=BIGTABLE_REGION
        
      2. Membatalkan tugas.

        gcloud dataflow jobs cancel JOB_ID --region=BIGTABLE_REGION
        

        Ganti JOB_ID dengan ID pekerjaan yang ditampilkan setelah perintah sebelumnya.

    Langkah selanjutnya