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

Diffuser des modifications avec Dataflow

Le connecteur Beam Bigtable vous permet d'utiliser Dataflow pour lire les enregistrements de modification des données Bigtable sans avoir à suivre ni à traiter les modifications de partition dans votre code, car le connecteur gère cette logique à votre place.

Ce document explique comment configurer et utiliser le connecteur Beam Bigtable pour lire un flux de modifications à l'aide d'un pipeline Dataflow. Avant de lire ce document, vous devez lire la présentation des flux de modifications et connaître Dataflow.

Alternatives à la création de votre propre pipeline

Si vous ne souhaitez pas créer votre propre pipeline Dataflow, vous pouvez utiliser l'une des options suivantes.

Vous pouvez utiliser un modèle Dataflow fourni par Google.

Vous pouvez également utiliser les exemples de code du tutoriel ou du démarrage rapide Bigtable comme point de départ pour votre code.

Assurez-vous que le code que vous générez utilise la version 26.14.0 ou ultérieure de google cloud libraries-bom.

Informations sur le connecteur

La méthode du connecteur Bigtable Beam, BigtableIO.readChangeStream, vous permet de lire un flux d'enregistrements de modification de données (ChangeStreamMutation) que vous pouvez traiter. Le connecteur Beam Bigtable est un composant du dépôt GitHub d'Apache Beam. Pour obtenir une description du code du connecteur, consultez les commentaires dans BigtableIO.java.

Vous devez utiliser le connecteur avec Beam version 2.48.0 ou ultérieure. Consultez la page Compatibilité des environnements d'exécution Apache Beam pour vous assurer que vous utilisez une version de Java compatible. Vous pouvez ensuite déployer un pipeline qui utilise le connecteur vers Dataflow, qui gère le provisionnement et la gestion des ressources, et facilite la scalabilité et la fiabilité du traitement des données de flux.

Pour en savoir plus sur le modèle de programmation Apache Beam, consultez la documentation Beam.

Regrouper des données sans heure de l'événement

Les enregistrements de modifications de données diffusés à l'aide du connecteur Beam Bigtable ne sont pas compatibles avec les fonctions Dataflow qui dépendent des heures d'événement.

Comme expliqué dans la section Réplication et repères, un repère bas peut ne pas avancer si la réplication de la partition n'a pas rattrapé le reste de l'instance. Lorsqu'un filigrane bas cesse d'avancer, le flux de modifications peut se bloquer.

Pour éviter que le flux ne se bloque, le connecteur Beam Bigtable génère toutes les données avec un code temporel de sortie de zéro. L'horodatage nul fait que Dataflow considère tous les enregistrements de modification de données comme des données tardives. Par conséquent, les fonctionnalités Dataflow qui dépendent des heures d'événement ne sont pas compatibles avec les flux de modifications Bigtable. Plus précisément, vous ne pouvez pas utiliser de fonctions de fenêtre, de déclencheurs basés sur le temps d'événement ni de délais basés sur le temps d'événement.

À la place, vous pouvez utiliser des GlobalWindows avec des déclencheurs non basés sur l'heure de l'événement pour regrouper ces données tardives dans des volets, comme illustré dans l'exemple du tutoriel. Pour en savoir plus sur les déclencheurs et les volets, consultez la section Déclencheurs du guide de programmation Beam.

Autoscaling

Le connecteur est compatible avec l'autoscaling Dataflow, qui est activé par défaut lorsque vous utilisez Runner v2 (obligatoire). L'algorithme d'autoscaling Dataflow tient compte du temps d'attente estimé du flux de modifications, qui peut être surveillé sur la page Surveillance de Dataflow dans la section Backlog. Utilisez l'option --maxNumWorkers lorsque vous déployez une tâche pour limiter le nombre de nœuds de calcul.

Pour effectuer le scaling manuel de votre pipeline au lieu d'utiliser l'autoscaling, consultez la section Effectuer le scaling manuel d'un pipeline de traitement par flux.

Limites

Notez les limites suivantes avant d'utiliser le connecteur Bigtable Beam avec Dataflow.

Exécuteur Dataflow v2

Le connecteur ne peut être exécuté qu'à l'aide de Dataflow Runner v2. Pour l'activer, spécifiez --experiments=use_runner_v2 dans vos arguments de ligne de commande. L'exécution avec l'exécuteur v1 entraîne l'échec de votre pipeline avec l'exception suivante:

java.lang.UnsupportedOperationException: BundleFinalizer unsupported by non-portable Dataflow

Instantanés

Le connecteur n'est pas compatible avec les instantanés Dataflow.

Doublons

Le connecteur Bigtable Beam diffuse les modifications pour chaque clé de ligne et chaque cluster dans l'ordre du code temporel de validation, mais comme il redémarre parfois à partir de dates antérieures dans le flux, il peut produire des doublons.

Avant de commencer

Avant d'utiliser le connecteur, vérifiez que vous remplissez les conditions préalables suivantes.

Configurer l'authentification

Pour utiliser les exemples Java de cette page dans un environnement de développement local, installez et initialisez gcloud CLI, puis configurez le service Identifiants par défaut de l'application à l'aide de vos identifiants utilisateur.

Install the Google Cloud CLI.
To initialize the gcloud CLI, run the following command:
```
gcloud init
```
If you're using a local shell, then create local authentication credentials for your user account:
```
gcloud auth application-default login
```
You don't need to do this if you're using Cloud Shell.

Pour en savoir plus, consultez Set up authentication for a local development environment.

Pour en savoir plus sur la configuration de l'authentification dans un environnement de production, consultez Set up Application Default Credentials for code running on Google Cloud.

Activer un flux de modifications

Vous devez activer un flux de modifications sur une table avant de pouvoir la lire. Vous pouvez également créer une table avec les flux de modifications activés.

Modifier la table des métadonnées du flux

Lorsque vous diffusez des modifications avec Dataflow, le connecteur Bigtable Beam crée une table de métadonnées appelée __change_stream_md_table par défaut. La table de métadonnées du flux de modifications gère l'état opérationnel du connecteur et stocke les métadonnées sur les enregistrements de modification des données.

Par défaut, le connecteur crée la table dans la même instance que la table en streaming. Pour que la table fonctionne correctement, le profil d'application de la table de métadonnées doit utiliser le routage à cluster unique et les transactions à ligne unique doivent être activées.

Pour en savoir plus sur le streaming de modifications à partir de Bigtable avec le connecteur Bigtable Beam, consultez la documentation BigtableIO.

Rôles requis

Pour obtenir les autorisations nécessaires pour lire un flux de modifications Bigtable à l'aide de Dataflow, demandez à votre administrateur de vous accorder les rôles IAM suivants.

Pour lire les modifications à partir de Bigtable, vous avez besoin du rôle suivant:

Administrateur Bigtable (roles/bigtable.admin) sur l'instance Bigtable contenant la table à partir de laquelle vous prévoyez de diffuser des modifications

Pour exécuter la tâche Dataflow, vous avez besoin des rôles suivants:

Développeur Dataflow (roles/dataflow.developer) sur le projet contenant vos ressources Cloud
Nœud de calcul Dataflow (roles/dataflow.worker) sur le projet contenant vos ressources Cloud
Administrateur des objets de l'espace de stockage (roles/storage.objectAdmin) sur les buckets Cloud Storage que vous prévoyez d'utiliser

Pour en savoir plus sur l'attribution de rôles, consultez la section Gérer les accès.

Vous pouvez également obtenir les autorisations requises via des rôles personnalisés ou d'autres rôles prédéfinis.

Ajouter le connecteur Bigtable Beam en tant que dépendance

Ajoutez du code semblable à la dépendance suivante à votre fichier pom.xml Maven. Vous devez disposer de la version 2.48.0 ou ultérieure.

<dependencies>
  <dependency>
    <groupId>org.apache.beam</groupId>
    <artifactId>beam-sdks-java-io-google-cloud-platform</artifactId>
    <version>VERSION</version>
  </dependency>
</dependencies>

Lire le flux de modifications

Pour créer un pipeline Dataflow permettant de lire vos enregistrements de modification de données, vous devez configurer le connecteur, puis ajouter des transformations et des récepteurs. Vous pouvez ensuite utiliser le connecteur pour lire des objets ChangeStreamMutation dans un pipeline Beam.

Les exemples de code de cette section, écrits en Java, montrent comment créer un pipeline et l'utiliser pour convertir des paires clé-valeur en chaîne. Chaque paire se compose d'une clé de ligne et d'un objet ChangeStreamMutation. Le pipeline convertit les entrées de chaque objet en chaîne séparée par des virgules.

Créer le pipeline

Cet exemple de code Java montre comment créer le pipeline:

BigtableOptions options =
    PipelineOptionsFactory.fromArgs(args).withValidation().as(BigtableOptions.class);
Pipeline p = Pipeline.create(options);

final Instant startTime = Instant.now();

p.apply(
        "Read Change Stream",
        BigtableIO.readChangeStream()
            .withProjectId(options.getBigtableProjectId())
            .withInstanceId(options.getBigtableInstanceId())
            .withTableId(options.getBigtableTableId())
            .withAppProfileId(options.getBigtableAppProfile())
            .withStartTime(startTime))
    .apply(
        "Flatten Mutation Entries",
        FlatMapElements.into(TypeDescriptors.strings())
            .via(ChangeStreamsHelloWorld::mutationEntriesToString))
    .apply(
        "Print mutations",
        ParDo.of(
            new DoFn<String, Void>() { // a DoFn as an anonymous inner class instance
              @ProcessElement
              public void processElement(@Element String mutation) {
                System.out.println("Change captured: " + mutation);
              }
            }));
p.run();

Traiter les enregistrements de modification des données

Cet exemple montre comment parcourir toutes les entrées d'un enregistrement de modification de données pour une ligne et appeler une méthode de conversion en chaîne en fonction du type d'entrée.

Pour obtenir la liste des types d'entrées qu'un enregistrement de modification de données peut contenir, consultez la section Contenu d'un enregistrement de modification de données.

static List<String> mutationEntriesToString(KV<ByteString, ChangeStreamMutation> mutationPair) {
  List<String> mutations = new ArrayList<>();
  String rowKey = mutationPair.getKey().toStringUtf8();
  ChangeStreamMutation mutation = mutationPair.getValue();
  MutationType mutationType = mutation.getType();
  for (Entry entry : mutation.getEntries()) {
    if (entry instanceof SetCell) {
      mutations.add(setCellToString(rowKey, mutationType, (SetCell) entry));
    } else if (entry instanceof DeleteCells) {
      mutations.add(deleteCellsToString(rowKey, mutationType, (DeleteCells) entry));
    } else if (entry instanceof DeleteFamily) {
      // Note: DeleteRow mutations are mapped into one DeleteFamily per-family
      mutations.add(deleteFamilyToString(rowKey, mutationType, (DeleteFamily) entry));
    } else {
      throw new RuntimeException("Entry type not supported.");
    }
  }
  return mutations;
}

Dans cet exemple, une entrée write est convertie:

private static String setCellToString(String rowKey, MutationType mutationType, SetCell setCell) {
  List<String> mutationParts =
      Arrays.asList(
          rowKey,
          mutationType.name(),
          "SetCell",
          setCell.getFamilyName(),
          setCell.getQualifier().toStringUtf8(),
          setCell.getValue().toStringUtf8());
  return String.join(",", mutationParts);
}

Dans cet exemple, une entrée de suppression de cellules est convertie:

private static String deleteCellsToString(
    String rowKey, MutationType mutationType, DeleteCells deleteCells) {
  String timestampRange =
      deleteCells.getTimestampRange().getStart() + "-" + deleteCells.getTimestampRange().getEnd();
  List<String> mutationParts =
      Arrays.asList(
          rowKey,
          mutationType.name(),
          "DeleteCells",
          deleteCells.getFamilyName(),
          deleteCells.getQualifier().toStringUtf8(),
          timestampRange);
  return String.join(",", mutationParts);
}

Dans cet exemple, la suppression d'une famille de colonnes est convertie:


private static String deleteFamilyToString(
    String rowKey, MutationType mutationType, DeleteFamily deleteFamily) {
  List<String> mutationParts =
      Arrays.asList(rowKey, mutationType.name(), "DeleteFamily", deleteFamily.getFamilyName());
  return String.join(",", mutationParts);
}

Surveiller

Les ressources suivantes de la console Google Cloud vous permettent de surveiller vos ressources Google Cloud lorsque vous exécutez un pipeline Dataflow pour lire un flux de modifications Bigtable:

Vérifiez en particulier les métriques suivantes:

Sur la page Monitoring (Surveillance) de Bigtable, vérifiez les métriques suivantes :
- Données sur l'utilisation du processeur par flux de modifications dans la métrique cpu_load_by_app_profile_by_method_by_table. Indique l'impact du flux de modifications sur l'utilisation du processeur de votre cluster.
- Flux de modifications de l'utilisation du stockage (octets) (change_stream_log_used_bytes)
Sur la page de surveillance de Dataflow, vérifiez la fraîcheur des données, qui indique la différence entre l'heure actuelle et le filigrane. Elle doit être d'environ deux minutes, avec des pics occasionnels d'une minute ou deux de plus. Si la métrique de fraîcheur des données est systématiquement supérieure à ce seuil, votre pipeline est probablement sous-ressourcé et vous devez ajouter des nœuds de calcul Dataflow. La fraîcheur des données n'indique pas si les enregistrements de modification des données sont traités lentement.
La métrique processing_delay_from_commit_timestamp_MEAN de Dataflow peut vous indiquer le temps de traitement moyen des enregistrements de modification des données sur la durée de vie de la tâche.

La métrique server/latencies Bigtable n'est pas utile lorsque vous surveillez un pipeline Dataflow qui lit un flux de modifications Bigtable, car elle reflète la durée de la requête de streaming, et non la latence de traitement des enregistrements de modification des données. Une latence élevée dans un flux de modifications ne signifie pas que les requêtes sont traitées lentement. Cela signifie que la connexion était ouverte pendant cette durée.