Google verwendet KI-Technologie, um Inhalte in Ihre bevorzugte Sprache zu übersetzen. KI-Übersetzungen können Fehler enthalten.

Bigtable-Änderungsstream verarbeiten

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

Es wird ein Beispieldatensatz für eine Musik-Streaming-Anwendung bereitgestellt. In dieser Anleitung erfassen Sie die Titel, die angehört werden, und erstellen dann eine Rangliste der Top 5 über einen bestimmten Zeitraum.

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

Ziele

In dieser Anleitung wird Folgendes beschrieben:

Eine Bigtable-Tabelle mit aktiviertem Änderungsstream erstellen.
Eine Pipeline in Dataflow bereitstellen, die den Änderungsstream transformiert und ausgibt.
Die Ergebnisse Ihrer Datenpipeline ansehen.

Kosten

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

Mit dem Preisrechner können Sie eine Kostenschätzung für Ihre voraussichtliche Nutzung vornehmen. Verwenden Sie den Preisrechner.

Neuen Google Cloud Nutzern vonsteht 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.

Hinweis

Melden Sie sich in Ihrem Google Cloud Konto an. Wenn Sie mit Google Cloudnoch nicht vertraut sind, erstellen Sie einfach ein Konto, um die Leistungsfähigkeit unserer Produkte in der Praxis sehen und bewerten zu können. Neukunden erhalten außerdem ein Guthaben von 300 $, um Arbeitslasten auszuführen, zu testen und bereitzustellen.

Installieren Sie die Google Cloud CLI. Initialisieren Sie die Google Cloud CLI nach der Installation mit dem folgenden Befehl:

gcloud init

Wenn Sie einen externen Identitätsanbieter (IdP) verwenden, müssen Sie sich zuerst mit Ihrer föderierten Identität in der gcloud CLI anmelden.

Erstellen oder wählen Sie ein Google Cloud Projekt aus.

Erforderliche Rollen zum Auswählen oder Erstellen eines Projekts

Projekt auswählen: Für die Auswahl eines Projekts ist keine bestimmte IAM-Rolle erforderlich. Sie können ein beliebiges Projekt auswählen, für das Ihnen eine Rolle zugewiesen wurde.
Projekt erstellen: Zum Erstellen eines Projekts benötigen Sie die Rolle „Projektersteller“ (roles/resourcemanager.projectCreator), die die resourcemanager.projects.create Berechtigung enthält. Rollen zuweisen.

Google Cloud Projekt erstellen:
```
gcloud projects create PROJECT_ID
```
Ersetzen Sie PROJECT_ID durch einen Namen für das Google Cloud Projekt, das Sie erstellen.
Wählen Sie das Google Cloud Projekt aus, das Sie erstellt haben:
```
gcloud config set project PROJECT_ID
```
Ersetzen Sie PROJECT_ID durch Ihren Google Cloud Projektnamen.

Prüfen Sie, ob die Abrechnung für Ihr Google Cloud Projekt aktiviert ist.

Aktivieren Sie die Dataflow API, die Cloud Bigtable API, die Cloud Bigtable Admin API und die Cloud Storage APIs:

Erforderliche Rollen zum Aktivieren von APIs

Zum Aktivieren von APIs benötigen Sie die Berechtigung serviceusage.services.enable. Wenn Sie das Projekt erstellt haben, haben Sie diese Berechtigung wahrscheinlich bereits über die Rolle „Inhaber“ (roles/owner). Andernfalls können Sie diese Berechtigung über die Rolle „Service Usage-Administrator“ (roles/serviceusage.serviceUsageAdmin) erhalten. Rollen zuweisen.

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

Installieren Sie die Google Cloud CLI. Initialisieren Sie die Google Cloud CLI nach der Installation mit dem folgenden Befehl:

gcloud init

Wenn Sie einen externen Identitätsanbieter (IdP) verwenden, müssen Sie sich zuerst mit Ihrer föderierten Identität in der gcloud CLI anmelden.

Erstellen oder wählen Sie ein Google Cloud Projekt aus.

Erforderliche Rollen zum Auswählen oder Erstellen eines Projekts

Projekt auswählen: Für die Auswahl eines Projekts ist keine bestimmte IAM-Rolle erforderlich. Sie können ein beliebiges Projekt auswählen, für das Ihnen eine Rolle zugewiesen wurde.
Projekt erstellen: Zum Erstellen eines Projekts benötigen Sie die Rolle „Projektersteller“ (roles/resourcemanager.projectCreator), die die resourcemanager.projects.create Berechtigung enthält. Rollen zuweisen.

Google Cloud Projekt erstellen:
```
gcloud projects create PROJECT_ID
```
Ersetzen Sie PROJECT_ID durch einen Namen für das Google Cloud Projekt, das Sie erstellen.
Wählen Sie das Google Cloud Projekt aus, das Sie erstellt haben:
```
gcloud config set project PROJECT_ID
```
Ersetzen Sie PROJECT_ID durch Ihren Google Cloud Projektnamen.

Prüfen Sie, ob die Abrechnung für Ihr Google Cloud Projekt aktiviert ist.

Aktivieren Sie die Dataflow API, die Cloud Bigtable API, die Cloud Bigtable Admin API und die Cloud Storage APIs:

Erforderliche Rollen zum Aktivieren von APIs

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

Aktualisieren und installieren Sie die cbt CLI .

gcloud components update
gcloud components install cbt

Umgebung vorbereiten

Code abrufen

Klonen Sie das Repository mit dem Beispielcode. Wenn Sie dieses Repository bereits heruntergeladen haben, rufen Sie die neueste Version ab.

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

Bucket erstellen

Erstellen Sie einen Cloud Storage-Bucket:

gcloud storage buckets create gs://BUCKET_NAME

Ersetzen Sie BUCKET_NAME durch einen Bucket-Namen der den Anforderungen für Bucket-Namen entspricht.

Bigtable-Instanz erstellen

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

Tabelle erstellen

Die Beispielanwendung erfasst die Titel, die Nutzer hören, und speichert die Wiedergabeereignisse in Bigtable. Erstellen Sie eine Tabelle mit aktiviertem Änderungsstream, einer Spaltenfamilie (cf) und einer Spalte (song), in der Nutzer-IDs für Zeilenschlüssel verwendet werden.

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 Projekts, das Sie verwenden
BIGTABLE_INSTANCE_ID: die ID der Instanz, die die neue Tabelle enthalten soll

Pipeline starten

Diese Pipeline transformiert den Änderungsstream so:

Liest den Änderungsstream
Ruft den Titelnamen ab
Gruppiert die Wiedergabeereignisse für den Titel in N-Sekunden-Fenstern
Zählt die Top 5 der Titel
Gibt die Ergebnisse aus

Führen Sie die Pipeline aus.

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.

Informationen zur Pipeline

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

Änderungsstream lesen

Der Code in diesem Beispiel konfiguriert den Quellstream mit den Parametern für die spezifische Bigtable-Instanz und -Tabelle.

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

    )

Titelnamen abrufen

Wenn ein Titel angehört wird, wird der Titelname in die Spaltenfamilie cf und den Spaltenqualifizierer song geschrieben. Der Code extrahiert den Wert aus der Änderungsstream-Mutation 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());
        }
      }
    }
  }
}

Top 5 der Titel zählen

Mit den integrierten Beam-Funktionen Count und Top.of können Sie die Top 5 der Titel im aktuellen Fenster abrufen.

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

Ergebnisse ausgeben

Diese Pipeline schreibt die Ergebnisse sowohl in die Standardausgabe als auch in Dateien. Für die Dateien werden die Schreibvorgänge in Gruppen von 10 Elementen oder in Ein-Minuten-Segmente unterteilt.

.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

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

Zu Dataflow
Klicken Sie auf den Job, dessen Name mit song-rank beginnt.
Klicken Sie unten auf dem Bildschirm auf Anzeigen , um das Logfeld zu öffnen.
Klicken Sie auf Worker-Logs , um die Ausgabelogs des Änderungsstreams zu beobachten.

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

Stream-Schreibvorgänge

Verwenden Sie die cbt CLI , um eine Reihe von Wiedergaben für verschiedene Nutzer in die song-rank Tabelle zu schreiben. Dies soll einige Minuten dauern, um zu simulieren, dass Titelwiedergaben 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

Lesen Sie die Ausgabe in Cloud Storage, um die beliebtesten Titel zu sehen.

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

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.

Projekt Google Cloud löschen:

gcloud projects delete PROJECT_ID

Einzelne Ressourcen löschen

Löschen Sie den Bucket und die Dateien.

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

Deaktivieren Sie den Änderungsstream in der Tabelle.

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

Löschen Sie die Tabelle song-rank.

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

Beenden Sie die Änderungsstream-Pipeline.
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.

Bigtable-Änderungsstream verarbeiten Mit Sammlungen den Überblick behalten Sie können Inhalte basierend auf Ihren Einstellungen speichern und kategorisieren.

Ziele

Kosten

Hinweis

Umgebung vorbereiten

Code abrufen

Bucket erstellen

Bigtable-Instanz erstellen

Tabelle erstellen

Pipeline starten

Informationen zur Pipeline

Änderungsstream lesen

Titelnamen abrufen

Top 5 der Titel zählen

Ergebnisse ausgeben

Pipeline ansehen

Stream-Schreibvorgänge

Ausgabe ansehen

Bereinigen

Projekt löschen

Einzelne Ressourcen löschen

Nächste Schritte

Bigtable-Änderungsstream verarbeiten