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 mendengarkan musik disediakan. Dalam tutorial ini, Anda akan melacak lagu yang didengarkan, lalu memberi peringkat lima lagu teratas selama jangka waktu tertentu.

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

Menyiapkan lingkungan

Mendapatkan kode

Buat clone repositori yang berisi kode contoh. Jika sebelumnya Anda telah 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

  • 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 dengarkan di Bigtable. Buat tabel dengan aliran perubahan yang diaktifkan yang memiliki satu family kolom (cf) dan satu kolom (lagu) serta 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

    Mulai pipeline

    Pipeline ini mengubah aliran perubahan dengan melakukan hal berikut:

    1. Membaca aliran perubahan
    2. Mendapatkan nama 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 berikut dari pipeline dapat membantu Anda memahami kode yang Anda jalankan.

    Membaca aliran 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 ditulis ke grup kolom cf dan penentu kolom song, sehingga kode mengekstrak nilai dari mutasi aliran perubahan dan menampilkannya ke langkah berikutnya dalam 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, file ini mengelompokkan penulisan ke dalam grup 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 song-rank.

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

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

    Penulisan streaming

    Gunakan CLI cbt untuk menulis sejumlah pendengaran lagu untuk berbagai pengguna ke tabel song-rank. Tindakan ini dirancang untuk menulis selama beberapa menit guna 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

    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}]