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

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.

Menginstal Google Cloud CLI.

Untuk initialize gcloud CLI, jalankan perintah berikut:

gcloud init

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.

Pastikan penagihan telah diaktifkan untuk project Google Cloud Anda.

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

Menginstal Google Cloud CLI.

Untuk initialize gcloud CLI, jalankan perintah berikut:

gcloud init

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.

Pastikan penagihan telah diaktifkan untuk project Google Cloud Anda.

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

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:

Membaca aliran data perubahan
Mendapatkan judul lagu
Mengelompokkan peristiwa mendengarkan lagu ke dalam jendela N-detik
Menghitung lima lagu teratas
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

Di konsol Google Cloud, buka halaman Dataflow.

Buka Dataflow
Klik tugas dengan nama yang diawali dengan peringkat lagu.
Di bagian bawah layar, klik Tampilkan untuk membuka panel log.
Klik Worker logs untuk memantau log output dari aliran data perubahan.

Catatan: Mungkin perlu waktu beberapa menit hingga log ini muncul saat tugas diinisialisasi.

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

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 berbagai arsitektur, tutorial, atau panduan memulai, menggunakan kembali project dapat membantu agar Anda tidak melampaui batas kuota project.

Menghapus project Google Cloud:

gcloud projects delete PROJECT_ID

Menghapus resource satu per satu

Hapus bucket dan file.

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

Nonaktifkan aliran data perubahan pada tabel.

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

Hapus tabel song-rank.

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

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.

Memproses aliran data perubahan Bigtable

Tujuan

Biaya

Sebelum memulai

Menyiapkan lingkungan

Mendapatkan kode

Membuat bucket

Membuat instance Bigtable

Membuat tabel

Memulai pipeline

Memahami pipeline

Membaca aliran data perubahan

Mendapatkan judul lagu

Menghitung lima lagu teratas

Menampilkan hasil

Melihat pipeline

Penulisan streaming

Melihat output

Pembersihan

Menghapus project

Menghapus resource satu per satu

Langkah selanjutnya