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 : de lire les enregistrements de modification de données Bigtable sans avoir à suivre traiter la partition des modifications apportées à vos 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, consultez la page Présentation des modifications flux et vous familiariser avec Dataflow.

Alternatives à la création de votre propre pipeline

Si vous ne souhaitez pas créer votre propre 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 Bigtable de démarrage rapide 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 de données. modifier les enregistrements (ChangeStreamMutation) que vous pouvez traiter. Le connecteur Bigtable Beam est du package GitHub d'Apache Beam un dépôt de clés. Pour une description du code de connecteur, consultez les commentaires disponibles à l'adresse BigtableIO.java

Vous devez utiliser le connecteur avec Beam version 2.48.0 ou ultérieure. Cochez la case Apache Beam compatibles avec l'environnement d'exécution. vous utilisez une version compatible de Java. Vous pouvez ensuite déployer un pipeline utilise le connecteur pour Dataflow, qui gère le provisionnement et la gestion ; de ressources, et contribue à l'évolutivité et à la fiabilité des flux de données en cours de traitement.

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 des données diffusés à l'aide du connecteur Bigtable Beam ne sont pas compatibles avec Dataflow. qui dépendent de l'heure des événements.

Comme expliqué dans la section Réplication et des filigranes, un un filigrane faible peut ne pas progresser si la réplication de la partition n'a pas été détectée jusqu'au reste de l'instance. Lorsqu'un faible filigrane cesse de progresser, entraîner le blocage du flux de modifications.

Pour éviter que le flux ne se bloque, le connecteur Bigtable Beam génère toutes les données dont le code temporel de sortie correspond à zéro. Avec un horodatage zéro, Dataflow considère toutes les données modifier les enregistrements pour être données tardives. Par conséquent, les fonctionnalités Dataflow qui dépendent de l'heure des événements ne sont pas est compatible avec les flux de modifications Bigtable. Plus précisément, vous ne pouvez pas utiliser les fonctions de fenêtrage ; déclencheurs liés à l'heure de l'événement ou minuteurs d'é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 Déclencheurs dans le guide de programmation Beam.

Autoscaling

Le connecteur est compatible avec Autoscaling Dataflow, activé par défaut lors de l'utilisation Runner v2 (obligatoire). L'algorithme d'autoscaling de Dataflow tient compte le backlog de flux de modifications estimé, qui peut être surveillé Surveillance Dataflow dans la section Backlog. Utilisez l'option --maxNumWorkers lorsque vous déployez un 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 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'avec Exécuteur Dataflow v2. Pour l'activer, spécifiez --experiments=use_runner_v2 dans votre ligne de commande . L'exécution avec Runner v1 entraîne l'échec de votre pipeline avec exception suivante:

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

Instantanés

Le connecteur n'est pas compatible Instantanés Dataflow.

Doublons

Le connecteur Bigtable Beam diffuse les modifications pour chaque clé de ligne et chaque dans l'ordre des codes temporels de commit, mais comme il peut redémarrer dans le flux, cela peut produire 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.

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 les 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 modifications de données ; d'enregistrements.

Par défaut, le connecteur crée la table dans la même instance que la table en cours de diffusion. Pour vous assurer que le tableau fonctionne correctement, le profil d'application pour la table de métadonnées doit utiliser un routage à cluster unique et ne comporter qu'une seule ligne transactions activées.

Pour en savoir plus sur les flux de modifications depuis Bigtable avec connecteur Bigtable Beam, consultez la section BigtableIO documentation.

Rôles requis

Pour obtenir les autorisations nécessaires pour lire une modification Bigtable à l'aide de Dataflow, demandez à votre administrateur de vous accorder 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 qui contient la table que vous prévoyez 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 Storage (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 rôles personnalisés ou autre 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 Maven pom.xml. La doit être 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 configurez le connecteur, puis vous ajoutez des transformations et des récepteurs. Ensuite, utilisez la méthode pour lire les objets ChangeStreamMutation dans un pipeline Beam.

Les exemples de code de cette section, écrits en Java, montrent comment créer un 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 chaque 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 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 Contenu d'un enregistrement de modifications 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 votre des ressources Google Cloud lorsque vous exécutez un pipeline Dataflow lire un flux de modifications Bigtable:

Vérifiez en particulier les métriques suivantes:

Sur la page Monitoring de Bigtable, vérifiez les points suivants : métriques: <ph type="x-smartling-placeholder">
- Données d'utilisation du processeur par flux de modifications dans la métrique cpu_load_by_app_profile_by_method_by_table Affiche le flux de modifications sur l'utilisation du processeur de votre cluster.
- Utilisation du stockage dans le flux de modifications (octets) (change_stream_log_used_bytes).
Sur la page de surveillance Dataflow, vérifiez les données data d'actualisation, qui indique la différence entre l'heure actuelle et la filigrane. Elle doit durer environ deux minutes, avec des pics occasionnels une ou deux minutes de plus. Si la métrique "Fraîcheur des données" est systématiquement plus élevée supérieur à ce seuil, votre pipeline manque probablement de ressources doit ajouter des nœuds de calcul Dataflow. La fraîcheur des données indiquer si les enregistrements de modification des données sont traités lentement.
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 du poste.

La métrique Bigtable server/latencies n'est pas utile pour surveiller un pipeline Dataflow qui lit le flux de modifications Bigtable, car il reflète la requête par flux et non la latence de traitement des enregistrements de modification des données. Latence élevée dans un flux de modifications ne signifie pas que les requêtes sont traitées lentement ; cela signifie la connexion a été ouverte pendant aussi longtemps.