Cette page a été traduite par l'API Cloud Translation.

Traiter un flux de modifications Bigtable

Ce tutoriel explique comment déployer un pipeline de données dans Dataflow pour un flux de modifications de base de données en temps réel provenant du flux de modifications d'une table Bigtable. La sortie du pipeline est écrite dans une série de fichiers sur Cloud Storage.

Un exemple de jeu de données pour une application d'écoute de musique est fourni. Dans ce tutoriel, vous allez suivre les titres écoutés, puis classer les cinq premiers sur une période donnée.

Ce tutoriel est destiné aux utilisateurs techniques qui savent écrire du code et déployer des pipelines de données dans Google Cloud.

Objectifs

Ce tutoriel vous explique comment effectuer les tâches suivantes :

Créez une table Bigtable avec un flux de modifications activé.
Déployez un pipeline sur Dataflow qui transforme et génère le flux de modifications.
Affichez les résultats de votre pipeline de données.

Coûts

Dans ce document, vous utilisez les composants facturables suivants de Google Cloud :

Obtenez une estimation des coûts en fonction de votre utilisation prévue à l'aide du simulateur de coût. Les nouveaux utilisateurs de Google Cloud peuvent bénéficier d'un essai gratuit.

Une fois que vous avez terminé les tâches décrites dans ce document, vous pouvez éviter de continuer à payer des frais en supprimant les ressources que vous avez créées. Pour en savoir plus, consultez la section Effectuer un nettoyage.

Avant de commencer

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.

Install the Google Cloud CLI.

To initialize the gcloud CLI, run the following command:

gcloud init

Create or select a Google Cloud 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.

Make sure that billing is enabled for your Google Cloud project.

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

Install the Google Cloud CLI.

To initialize the gcloud CLI, run the following command:

gcloud init

Create or select a Google Cloud 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.

Make sure that billing is enabled for your Google Cloud project.

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

Mettez à jour et installez la CLI cbt.

gcloud components update
gcloud components install cbt

Préparer l'environnement

Obtenir le code

Clonez le dépôt contenant l'exemple de code. Si vous avez déjà téléchargé ce dépôt, récupérez la dernière version.

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

Créer un 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.

Créer une instance Bigtable

Vous pouvez utiliser une instance existante pour ce tutoriel ou en créer une avec les configurations par défaut dans une région proche de chez vous.

Créer une table

L'application exemple suit les titres que les utilisateurs écoutent et stocke les événements d'écoute dans Bigtable. Créez une table avec un flux de modifications activé qui comporte une famille de colonnes (cf) et une colonne (chanson) et qui utilise des ID utilisateur pour les clés de ligne.

Créez la table.

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

Remplacez les éléments suivants :

PROJECT_ID: ID du projet que vous utilisez
BIGTABLE_INSTANCE_ID: ID de l'instance qui contiendra la nouvelle table

Démarrer le pipeline

Ce pipeline transforme le flux de modifications en procédant comme suit:

Lit le flux de modifications
Récupère le nom de la chanson
Regroupe les événements d'écoute de titres dans des fenêtres de N secondes
Compte les cinq titres les plus écoutés
Affiche les résultats

Exécutez le 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"

Remplacez BIGTABLE_REGION par l'ID de la région dans laquelle se trouve votre instance Bigtable, par exemple us-east5.

Comprendre le pipeline

Les extraits de code suivants du pipeline peuvent vous aider à comprendre le code que vous exécutez.

Lire le flux de modifications

Le code de cet exemple configure le flux source avec les paramètres de l'instance et de la table Bigtable spécifiques.

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

    )

Obtenir le nom de la chanson

Lorsqu'un titre est écouté, son nom est écrit dans la famille de colonnes cf et le qualificatif de colonne song. Le code extrait donc la valeur de la mutation du flux de modifications et la renvoie à l'étape suivante du 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());
        }
      }
    }
  }
}

Compter les cinq titres les plus écoutés

Vous pouvez utiliser les fonctions Beam intégrées Count et Top.of pour obtenir les cinq titres les plus populaires dans la fenêtre actuelle.

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

Afficher les résultats

Ce pipeline écrit les résultats dans la sortie standard et dans des fichiers. Pour les fichiers, il regroupe les écritures en groupes de 10 éléments ou de segments d'une minute.

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

Afficher le pipeline

Dans la console Google Cloud, accédez à la page Dataflow.

Accéder à Dataflow
Cliquez sur la tâche dont le nom commence par song-rank.
En bas de l'écran, cliquez sur Afficher pour ouvrir le panneau des journaux.
Cliquez sur Journaux des nœuds de calcul pour surveiller les journaux de sortie du flux de modifications.

Remarque :L'apparition du journal peut prendre quelques minutes pendant l'initialisation de la tâche.

Écritures en flux

Utilisez la CLI cbt pour enregistrer un nombre d'écoutes de titres pour différents utilisateurs dans la table song-rank. Il est conçu pour écrire sur plusieurs minutes afin de simuler l'écoute de titres en streaming au fil du temps.

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

Consulter le résultat

Lisez la sortie dans Cloud Storage pour afficher les titres les plus populaires.

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

Exemple de résultat :

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

Effectuer un nettoyage

Pour éviter que les ressources utilisées lors de ce tutoriel soient facturées sur votre compte Google Cloud, supprimez le projet contenant les ressources, ou conservez le projet et supprimez les ressources individuelles.

Supprimer le projet

Attention : La suppression d'un projet entraîne les effets décrits ci-dessous :

Tout le contenu du projet est supprimé. Si vous avez utilisé un projet existant pour les tâches décrites dans ce document et que vous le supprimez, vous supprimerez également tout autre travail effectué dans le projet.
Les ID de projets personnalisés sont perdus. Lorsque vous avez créé ce projet, vous avez peut-être créé un ID de projet personnalisé que vous souhaitez utiliser à l'avenir. Pour conserver les URL qui utilisent l'ID de projet, telle qu'une URL appspot.com, supprimez les ressources sélectionnées dans le projet au lieu de supprimer l'ensemble du projet.

Si vous envisagez d'explorer plusieurs architectures, tutoriels et guides de démarrage rapide, réutiliser des projets peut vous aider à ne pas dépasser les limites de quotas des projets.

Delete a Google Cloud project:

gcloud projects delete PROJECT_ID

Supprimer des ressources individuelles

Supprimez le bucket et les fichiers.

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

Désactivez le flux de modifications sur la table.

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

Supprimez la table song-rank.

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

Arrêtez le pipeline du flux de modifications.
1. Répertoriez les tâches pour obtenir l'ID de la tâche.
```
gcloud dataflow jobs list --region=BIGTABLE_REGION
```
2. Annulez la tâche.
```
gcloud dataflow jobs cancel JOB_ID --region=BIGTABLE_REGION
```
  Remplacez JOB_ID par l'ID de tâche affiché après la commande précédente.

Traiter un flux de modifications Bigtable

Objectifs

Coûts

Avant de commencer

Préparer l'environnement

Obtenir le code

Créer un bucket

Créer une instance Bigtable

Créer une table

Démarrer le pipeline

Comprendre le pipeline

Lire le flux de modifications

Obtenir le nom de la chanson

Compter les cinq titres les plus écoutés

Afficher les résultats

Afficher le pipeline

Écritures en flux

Consulter le résultat

Effectuer un nettoyage

Supprimer le projet

Supprimer des ressources individuelles

Étape suivante