Bigtable-Änderungsstream verarbeiten

---

In dieser Anleitung erfahren Sie, wie Sie eine Datenpipeline in Dataflow für einen Echtzeitstream von Datenbankänderungen bereitstellen, der aus dem Änderungsstream einer Bigtable-Tabelle stammt. Die Ausgabe der Pipeline wird in eine Reihe von Dateien in Cloud Storage geschrieben.

Es wird ein Beispieldatensatz für eine Musik-App bereitgestellt. In dieser Anleitung erfassen Sie die angehörten Titel und sortieren die Top 5 über einen bestimmten Zeitraum.

Diese Anleitung richtet sich an technische Nutzer, die mit dem Schreiben von Code und dem Bereitstellen von Datenpipelines in Google Cloud vertraut sind.

Lernziele

In dieser Anleitung wird Folgendes beschrieben:

• Erstellen Sie eine Bigtable-Tabelle mit aktiviertem Änderungsstream.
• Stellen Sie eine Pipeline in Dataflow bereit, die den Änderungsstream transformiert und ausgibt.
• Sehen Sie sich die Ergebnisse Ihrer Datenpipeline an.

Kosten

In diesem Dokument verwenden Sie die folgenden kostenpflichtigen Komponenten von Google Cloud:

• Bigtable
• Compute Engine
• Dataflow
• Cloud Storage

Mit dem Preisrechner können Sie eine Kostenschätzung für Ihre voraussichtliche Nutzung vornehmen. Neuen Google Cloud-Nutzern steht möglicherweise eine kostenlose Testversion zur Verfügung.

Nach Abschluss der in diesem Dokument beschriebenen Aufgaben können Sie weitere Kosten vermeiden, indem Sie die erstellten Ressourcen löschen. Weitere Informationen finden Sie unter Bereinigen.

Hinweise

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. Aktualisieren und installieren Sie die cbt CLI.

    gcloud components update
    gcloud components install cbt

Umgebung vorbereiten

Code abrufen

Klonen Sie das Repository, das den Beispielcode enthält. Wenn Sie dieses Repository bereits heruntergeladen haben, können Sie mit „Pull“ die neueste Version abrufen.

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

Bucket erstellen

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.

Bigtable-Instanz erstellen

Sie können für diese Anleitung eine vorhandene Instanz verwenden oder eine Instanz mit den Standardkonfigurationen in einer Region in Ihrer Nähe erstellen.

Tabelle erstellen

Die Beispielanwendung erfasst die Songs, die sich Nutzer anhören, und speichert die Wiedergabeereignisse in Bigtable. Erstellen Sie eine Tabelle mit einem aktivierten Änderungsstream, die eine Spaltenfamilie (cf) und eine Spalte (song) enthält und Nutzer-IDs als Zeilenschlüssel verwendet.

Erstellen Sie die Tabelle.

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

Ersetzen Sie Folgendes:

• PROJECT_ID: die ID des verwendeten Projekts
• BIGTABLE_INSTANCE_ID: Die ID der Instanz, die die neue Tabelle enthalten soll.

Pipeline starten

In dieser Pipeline wird der Änderungsstream so transformiert:

1. Liest den Änderungsstream
2. Ruft den Namen des Titels ab.
3. Gruppiert die Wiedergabeereignisse des Titels in Zeitfenstern von N Sekunden.
4. Die fünf meistgespielten Titel werden gezählt.
5. Gibt die Ergebnisse aus

Pipeline ausführen.

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"

Ersetzen Sie BIGTABLE_REGION durch die ID der Region, in der sich Ihre Bigtable-Instanz befindet, z. B. us-east5.

Pipeline

Die folgenden Code-Snippets aus der Pipeline können Ihnen helfen, den Code zu verstehen, den Sie ausführen.

Änderungsstream lesen

Im Code in diesem Beispiel wird der Quellstream mit den Parametern für die jeweilige Bigtable-Instanz und ‑Tabelle konfiguriert.

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

    )

Songtitel abrufen

Wenn ein Song angehört wird, wird der Songname in die Spaltenfamilie cf und den Spaltenqualifizierer song geschrieben. Der Code extrahiert den Wert aus der Mutation des Änderungsstreams und gibt ihn an den nächsten Schritt der Pipeline aus.

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());
        }
      }
    }
  }
}

Die fünf meistgespielten Songs

Mit den integrierten Beam-Funktionen Count und Top.of kannst du die fünf beliebtesten Titel im aktuellen Fenster abrufen.

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

Ergebnisse ausgeben

Bei dieser Pipeline werden die Ergebnisse sowohl in Standard-Out als auch in Dateien geschrieben. Bei den Dateien werden die Schreibvorgänge in Gruppen von 10 Elementen oder einminütigen Segmenten zusammengefasst.

.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()
);

Pipeline ansehen

1. Rufen Sie in der Google Cloud Console die Seite Dataflow auf.

   Zu Dataflow

2. Klicke auf den Job, dessen Name mit song-rank beginnt.

3. Klicken Sie unten auf dem Bildschirm auf Anzeigen, um das Logfeld zu öffnen.

4. Klicken Sie auf Worker-Logs, um die Ausgabeprotokolle des Änderungsstreams zu überwachen.

   Hinweis: Es kann einige Minuten dauern, bis das Protokoll angezeigt wird, während der Job initialisiert wird.

Streamschreibvorgänge

Verwenden Sie die cbt-Befehlszeile, um die Anzahl der Songwiedergaben für verschiedene Nutzer in die Tabelle song-rank zu schreiben. Die Daten werden über einen Zeitraum von mehreren Minuten geschrieben, um die Wiedergabe von Songs zu simulieren, die im Laufe der Zeit gestreamt werden.

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

Ausgabe ansehen

Rufe die Ausgabe in Cloud Storage auf, um die beliebtesten Songs zu sehen.

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

Hinweis: Es kann einige Minuten dauern, bis die Datei angezeigt wird, während die Pipeline die Daten verarbeitet. Sehen Sie sich die Dataflow-Worker-Logs an, um den Status zu prüfen.

Beispielausgabe:

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

Bereinigen

Damit Ihrem Google Cloud-Konto die in dieser Anleitung verwendeten Ressourcen nicht in Rechnung gestellt werden, löschen Sie entweder das Projekt, das die Ressourcen enthält, oder Sie behalten das Projekt und löschen die einzelnen Ressourcen.

Projekt löschen

Achtung: Das Löschen von Projekten hat folgende Auswirkungen:

• Alle Inhalte des Projekts werden gelöscht. Wenn Sie für die Aufgaben in diesem Dokument ein bereits bestehendes Projekt verwendet haben und dieses löschen, werden auch alle anderen im Rahmen des Projekts erstellten Daten gelöscht.
• Benutzerdefinierte Projekt-IDs gehen verloren. Beim Erstellen dieses Projekts haben Sie möglicherweise eine benutzerdefinierte Projekt-ID erstellt, die Sie weiterhin verwenden möchten. Damit die URLs, die die Projekt-ID nutzen, z. B. eine appspot.com-URL, erhalten bleiben, sollten Sie ausgewählte Ressourcen innerhalb des Projekts löschen, statt das gesamte Projekt.

Wenn Sie mehrere Architekturen, Anleitungen und Kurzanleitungen durcharbeiten möchten, können Sie die Überschreitung von Projektkontingenten verhindern, indem Sie Projekte wiederverwenden.

Delete a Google Cloud project:

gcloud projects delete PROJECT_ID

Einzelne Ressourcen löschen

1. Löschen Sie den Bucket und die Dateien.

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

2. Deaktivieren Sie den Änderungsstream für die Tabelle.

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

3. Löschen Sie die Tabelle song-rank.

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

4. Halten Sie die Änderungsstream-Pipeline an.

   1. Listen Sie die Jobs auf, um die Job-ID zu erhalten.

      gcloud dataflow jobs list --region=BIGTABLE_REGION
      

   2. Brechen Sie den Job ab.

      gcloud dataflow jobs cancel JOB_ID --region=BIGTABLE_REGION
      

      Ersetzen Sie JOB_ID durch die Job-ID, die nach dem vorherigen Befehl angezeigt wird.

Nächste Schritte

• Lesen Sie die Dokumentation zu Änderungsstreams.
• Beispielcode herunterladen