Lire un flux de modifications avec Java

La bibliothèque cliente Cloud Bigtable pour Java fournit des méthodes de bas niveau pour traiter les enregistrements de modification de données. Toutefois, dans la plupart des cas, nous vous recommandons de transmettre les modifications avec Dataflow plutôt que d'utiliser les méthodes décrites sur cette page, car Dataflow gère les partitions et les fusions pour vous.

Avant de commencer

Avant de lire un flux de modifications avec Java, assurez-vous de connaître la présentation des flux de modifications. Complétez ensuite 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.

  1. Install the Google Cloud CLI.

  2. To initialize the gcloud CLI, run the following command: 

    gcloud init

  3. 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 un flux de modifications activé.

Rôles requis

Pour obtenir les autorisations nécessaires pour lire un flux de modifications Bigtable, demandez à votre administrateur de vous accorder le rôle IAM suivant.

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

Ajouter la bibliothèque cliente Java en tant que dépendance

Ajoutez un code semblable au code suivant à votre fichier pom.xml Maven. Remplacez VERSION par la version de la bibliothèque cliente que vous utilisez. La version doit être 2.21.0 ou ultérieure.

<dependencies>
  <dependency>
    <groupId>com.google.cloud</groupId>
    <artifactId>google-cloud-bigtable</artifactId>
    <version>VERSION</version>
  </dependency>
</dependencies>

Déterminer les partitions de la table

Pour commencer à envoyer des requêtes ReadChangeStream, vous devez connaître les partitions de votre table. Cela peut être déterminé à l'aide de la méthode GenerateInitialChangeStreamPartitions. L'exemple suivant montre comment utiliser cette méthode pour obtenir un flux de ByteStringRanges représentant chaque partition de la table. Chaque ByteStringRange contient la clé de début et de fin d'une partition.

ServerStream<ByteStringRange> partitionStream =
    client.generateInitialChangeStreamPartitions("MyTable");

Traiter les modifications pour chaque partition

Vous pouvez ensuite traiter les modifications pour chaque partition à l'aide de la méthode ReadChangeStream. Voici un exemple d'ouverture d'un flux pour une partition, à partir de l'heure actuelle.

ReadChangeStreamQuery query =
    ReadChangeStreamQuery.create("MyTable")
        .streamPartition(partition)
        .startTime(Instant.now());
ServerStream<ChangeStreamRecord> changeStream = client.readChangeStream(query);

ReadChangeStreamQuery accepte les arguments suivants:

  • Partition de flux (obligatoire) : partition à partir de laquelle diffuser les modifications
  • Choisissez l'une des options suivantes :
    • Heure de début : code temporel de validation à partir duquel le traitement des modifications commence
    • Jetons de continuation : jetons représentant une position à partir de laquelle reprendre le streaming
  • Heure de fin (facultatif) : code temporel de validation pour arrêter le traitement des modifications lorsqu'il est atteint. Si vous ne fournissez pas de valeur, le flux continue de lire.
  • Durée du battement de cœur (facultatif) : fréquence des messages de battement de cœur en l'absence de nouveaux changements (par défaut, cinq secondes)

Modifier le format d'enregistrement du flux

Un enregistrement de flux de modifications renvoyé est l'un des trois types de réponses suivants:

  • ChangeStreamMutation : message représentant un enregistrement de modification de données.

  • CloseStream : message indiquant que le client doit arrêter de lire à partir du flux.

    • État : indique le motif de la fermeture du flux. Au choix :
      • OK : l'heure de fin a été atteinte pour la partition donnée
      • OUT_OF_RANGE : la partition donnée n'existe plus, ce qui signifie que des fractionnements ou des fusions ont été effectués sur cette partition. Une nouvelle requête ReadChangeStream doit être créée pour chaque nouvelle partition.
    • NewPartitions : fournit les informations de partitionnement mises à jour sur les réponses OUT_OF_RANGE.
    • ChangeStreamContinuationTokens : liste des jetons utilisés pour reprendre de nouvelles requêtes ReadChangeStream à partir de la même position. Une par NewPartition.

  • Heartbeat : message périodique contenant des informations pouvant être utilisées pour vérifier l'état du flux.

    • EstimatedLowWatermark : estimation du repère bas pour la partition donnée
    • ContinuationToken : jeton permettant de reprendre la diffusion de la partition donnée à partir de la position actuelle.

Contenu des enregistrements de modification des données

Pour en savoir plus sur les enregistrements de flux de modifications, consultez la section Contenu d'un enregistrement de modification des données.

Gérer les modifications des partitions

Lorsque les partitions d'une table changent, les requêtes ReadChangeStream renvoient un message CloseStream contenant les informations requises pour reprendre le streaming à partir de la ou des nouvelles partitions.

Pour une division, il contient plusieurs nouvelles partitions et un ContinuationToken correspondant pour chaque partition. Pour reprendre la diffusion des nouvelles partitions à partir de la même position, vous devez envoyer une nouvelle requête ReadChangeStream pour chaque nouvelle partition avec son jeton correspondant.

Par exemple, si vous diffusez en continu la partition [A,C) et qu'elle se divise en deux partitions, [A,B) et [B,C), vous pouvez vous attendre à la séquence d'événements suivante:

ReadChangeStream(streamPartition = ByteStringRange(A, C)) receives:
CloseStream(
    Status = OUT_OF_RANGE,
    NewPartitions = List(ByteStringRange(A, B), ByteStringRange(B, C))
    ChangeStreamContinuationTokens = List(foo, bar)
)

Pour reprendre la diffusion en continu de chaque partition à partir de la même position, envoyez les requêtes ReadChangeStreamQuery suivantes:

ReadChangeStreamQuery queryAB =
    ReadChangeStreamQuery.create("myTable")
        .streamPartition(ByteStringRange(A, B))
        .continuationTokens(List.of(foo));

ReadChangeStreamQuery queryBC =
    ReadChangeStreamQuery.create("myTable")
        .streamPartition(ByteStringRange(B, C))
        .continuationTokens(List.of(bar));

Pour une fusion, pour reprendre à partir de la même partition, vous devez envoyer une nouvelle requête ReadChangeStream contenant chaque jeton des partitions fusionnées.

Par exemple, si vous lisez en streaming deux partitions, [A,B) et [B,C), et qu'elles fusionnent dans la partition [A,C), vous pouvez vous attendre à la séquence d'événements suivante:

ReadChangeStream(streamPartition = ByteStringRange(A, B)) receives:
CloseStream(
    Status = OUT_OF_RANGE,
    NewPartitions = List(ByteStringRange(A, C)),
    ChangeStreamContinuationTokens = List(foo)
)

ReadChangeStream(streamPartition = ByteStringRange(B, C)) receives:
CloseStream(
    Status = OUT_OF_RANGE,
    NewPartitions = List(ByteStringRange(A, C)),
    ChangeStreamContinuationTokens = List(bar)
)

Pour reprendre la partition de streaming [A, C) à partir de la même position, envoyez un ReadChangeStreamQuery comme suit:

ReadChangeStreamQuery query =
    ReadChangeStreamQuery.create("myTable")
        .streamPartition(ByteStringRange(A, C))
        .continuationTokens(List.of(foo, bar));

Étape suivante