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

Diffuser les modifications avec Dataflow

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

Ce document explique comment configurer et utiliser le connecteur Bigtable Beam 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 vous familiariser avec 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 google cloud libraries-bom version 26.14.0 ou ultérieure.

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 Bigtable Beam 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. Vérifiez la compatibilité de l'environnement 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 à Dataflow, qui gère le provisionnement et la gestion des ressources, et contribue à l'évolutivité et à la fiabilité du traitement des données par flux.

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

Regrouper des données sans heure d'événement

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

Comme expliqué dans la section Réplication et filigranes, un filigrane faible peut ne pas progresser si la réplication de la partition n'a pas rattrapé le reste de l'instance. Lorsqu'un filigrane faible cesse de progresser, le flux de modifications peut se bloquer.

Pour éviter le blocage du flux, le connecteur Bigtable Beam génère toutes les données avec un code temporel de sortie de zéro. Avec l'horodatage zéro, 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êtrage, de déclencheurs basés sur l'heure de l'événement ni de minuteurs à l'heure de l'événement.

À la place, vous pouvez utiliser GlobalWindows avec des déclencheurs d'heure sans é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 de Dataflow, qui est activé par défaut lors de l'utilisation de Runner v2 (obligatoire). L'algorithme d'autoscaling de Dataflow prend en compte l'estimation du trafic en attente du flux de modifications, qui peut être surveillée sur la page Surveillance de Dataflow dans la section Backlog. Utilisez l'option --maxNumWorkers lors du déploiement d'une tâche pour limiter le nombre de nœuds de calcul.

Pour procéder au scaling manuel de votre pipeline au lieu d'utiliser l'autoscaling, consultez la page Effectuer le scaling manuel d'un pipeline de flux de données.

Limites

Prenez note des 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 des codes temporels de commit. Toutefois, comme il redémarre parfois à des intervalles précédents du flux, il peut générer des doublons.

Avant de commencer

Avant d'utiliser le connecteur, vous devez remplir 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.

Installez Google Cloud CLI.
Pour initialiser gcloudCLI, exécutez la commande suivante :
```
gcloud init
```

Créez des identifiants d'authentification locaux pour votre compte Google :
```
gcloud auth application-default login
```

Pour en savoir plus, consultez les sections sur Configurer l'authentification pour un environnement de développement local.

Pour en savoir plus sur la configuration de l'authentification dans un environnement de production, consultez Configurer le service Identifiants par défaut de l'application pour le code exécuté sur Google Cloud.

Activer un flux de modifications

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

Modifier la table des métadonnées de flux

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

Par défaut, le connecteur crée la table dans la même instance que la table diffusée en continu. Pour garantir le bon fonctionnement de la table, le profil d'application de la table des métadonnées doit utiliser un routage à cluster unique et les transactions à ligne unique doivent être activées.

Pour en savoir plus sur les flux de modifications de Bigtable avec le connecteur Bigtable Beam, consultez la documentation de 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 attribuer les rôles IAM suivants.

Pour lire les modifications depuis Bigtable, vous devez disposer de ce rôle:

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

Pour exécuter le job 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 un code semblable à la dépendance suivante à votre fichier Maven pom.xml. Vous devez utiliser la version 2.48.0 ou une version 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 afin de lire vos enregistrements de modification de données, vous devez configurer le connecteur, puis ajouter des transformations et des récepteurs. Vous utiliserez ensuite le connecteur pour lire les objets ChangeStreamMutation dans un pipeline Beam.

Les exemples de code de cette section, écrits en Java, montrent comment construire un pipeline et l'utiliser pour convertir des paires clé-valeur en une 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 une 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 lire en boucle toutes les entrées d'un enregistrement de modification de données pour une ligne et comment 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 de Bigtable, vérifiez les metrics suivantes :
- Utilisation du processeur par flux de modifications dans la métrique cpu_load_by_app_profile_by_method_by_table. Affiche l'impact du flux de modifications sur l'utilisation du processeur de votre cluster.
- Utilisation du stockage des flux de modifications (octets) (change_stream_log_used_bytes).
Sur la page de surveillance Dataflow, vérifiez la fraîcheur des données, qui indique la différence entre l'heure actuelle et le filigrane. Elle doit durer environ deux minutes, avec des pics occasionnels d'une ou deux minutes de plus. Si la métrique de fraîcheur des données est systématiquement supérieure à ce seuil, votre pipeline manque probablement de ressources et vous devriez ajouter davantage de nœuds de calcul Dataflow. La fraîcheur des données n'indique pas si les enregistrements de modification de données sont traités lentement.
La métrique processing_delay_from_commit_timestamp_MEAN Dataflow peut vous indiquer le temps de traitement moyen des enregistrements de modification de données au cours de la durée de vie de la tâche.

La métrique Bigtable server/latencies 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 en flux continu, et non la latence du traitement des enregistrements de modification des données. Une latence élevée dans un flux de modification ne signifie pas que les requêtes sont traitées lentement, mais que la connexion a été ouverte pendant cette longue durée.