Connecteur Dataflow pour Bigtable

Le connecteur Cloud Dataflow pour Cloud Bigtable permet d'utiliser Bigtable dans un pipeline Cloud Dataflow. Vous pouvez utiliser le connecteur pour les opérations par lots et pour les opérations de streaming.

Le connecteur est écrit dans Java et est construit sur le client Bigtable HBase pour Java. Il est compatible avec le SDK Dataflow 2.x pour Java, basé sur Apache Beam. Le code source du connecteur se trouve sur GitHub, dans le dépôt googleapis/java-bigtable-hbase.

Cette page décrit l'utilisation des transformations Read et Write avec le connecteur Cloud Dataflow. Vous pouvez également consulter la documentation complète de l'API pour en savoir plus sur le connecteur Cloud Dataflow.

Ajouter le connecteur à un projet Maven

Pour ajouter le connecteur Cloud Dataflow à un projet Maven, ajoutez l'artefact Maven au fichier pom.xml, sous la forme d'une dépendance :

pom.xml

Afficher sur GitHub

<dependency>
  <groupId>com.google.cloud.bigtable</groupId>
  <artifactId>bigtable-hbase-beam</artifactId>
  <version>1.25.0</version>
</dependency>

Définir la configuration Bigtable

Créez une interface d'options pour permettre des entrées pour le fonctionnement de votre pipeline :

HelloWorldWrite.java

Afficher sur GitHub

public interface BigtableOptions extends DataflowPipelineOptions {
  @Description("The Bigtable project ID, this can be different than your Dataflow project")
  @Default.String("bigtable-project")
  String getBigtableProjectId();

  void setBigtableProjectId(String bigtableProjectId);

  @Description("The Bigtable instance ID")
  @Default.String("bigtable-instance")
  String getBigtableInstanceId();

  void setBigtableInstanceId(String bigtableInstanceId);

  @Description("The Bigtable table ID in the instance.")
  @Default.String("bigtable-table")
  String getBigtableTableId();

  void setBigtableTableId(String bigtableTableId);
}

Lorsque vous lisez ou écrivez dans Bigtable, vous devez fournir un objet de configuration CloudBigtableConfiguration. Cet objet indique l'ID de projet et l'ID d'instance de votre table, ainsi que le nom de la table elle-même :

HelloWorldWrite.java

Afficher sur GitHub

CloudBigtableTableConfiguration bigtableTableConfig =
    new CloudBigtableTableConfiguration.Builder()
        .withProjectId(options.getBigtableProjectId())
        .withInstanceId(options.getBigtableInstanceId())
        .withTableId(options.getBigtableTableId())
        .build();

Pour la lecture, fournissez un objet de configuration CloudBigtableScanConfiguration, qui vous permet de spécifier un objet Apache HBase Scan qui limite et filtre les résultats d'une lecture. Pour en savoir plus, consultez la section Lire depuis Bigtable.

Lire des données depuis Bigtable

Pour lire depuis une table Bigtable, vous devez appliquer une transformation Read au résultat d'une opération CloudBigtableIO.read. La transformation Read renvoie une collection PCollectiond'objets HBase Result, où chaque élément de la PCollection représente une ligne individuelle de la table.

HelloWorldRead.java

Afficher sur GitHub

p.apply(Read.from(CloudBigtableIO.read(config)))
    .apply(
        ParDo.of(
            new DoFn<Result, Void>() {
              @ProcessElement
              public void processElement(@Element Result row, OutputReceiver<Void> out) {
                System.out.println(Bytes.toString(row.getRow()));
              }
            }));

Par défaut, une opération CloudBigtableIO.read renvoie toutes les lignes de votre table. Vous pouvez utiliser un objet HBase Scan pour limiter la lecture à une plage de clés de ligne de la table, ou pour appliquer des filtres aux résultats de la lecture. Pour utiliser un objet Scan, incluez-le dans votre objet CloudBigtableScanConfiguration.

Par exemple, vous pouvez ajouter un élément Scan qui renvoie seulement la première paire clé-valeur de chaque ligne de la table, ce qui est utile lorsque vous comptez le nombre de lignes de la table :

HelloWorldRead.java

Afficher sur GitHub

import com.google.cloud.bigtable.beam.CloudBigtableIO;
import com.google.cloud.bigtable.beam.CloudBigtableScanConfiguration;
import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.Read;
import org.apache.beam.sdk.options.Default;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.FirstKeyOnlyFilter;
import org.apache.hadoop.hbase.util.Bytes;

public class HelloWorldRead {
  public static void main(String[] args) {
    BigtableOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(BigtableOptions.class);
    Pipeline p = Pipeline.create(options);

    Scan scan = new Scan();
    scan.setCacheBlocks(false);
    scan.setFilter(new FirstKeyOnlyFilter());

    CloudBigtableScanConfiguration config =
        new CloudBigtableScanConfiguration.Builder()
            .withProjectId(options.getBigtableProjectId())
            .withInstanceId(options.getBigtableInstanceId())
            .withTableId(options.getBigtableTableId())
            .withScan(scan)
            .build();

    p.apply(Read.from(CloudBigtableIO.read(config)))
        .apply(
            ParDo.of(
                new DoFn<Result, Void>() {
                  @ProcessElement
                  public void processElement(@Element Result row, OutputReceiver<Void> out) {
                    System.out.println(Bytes.toString(row.getRow()));
                  }
                }));

    p.run().waitUntilFinish();
  }

  public interface BigtableOptions extends DataflowPipelineOptions {
    @Description("The Bigtable project ID, this can be different than your Dataflow project")
    @Default.String("bigtable-project")
    String getBigtableProjectId();

    void setBigtableProjectId(String bigtableProjectId);

    @Description("The Bigtable instance ID")
    @Default.String("bigtable-instance")
    String getBigtableInstanceId();

    void setBigtableInstanceId(String bigtableInstanceId);

    @Description("The Bigtable table ID in the instance.")
    @Default.String("bigtable-table")
    String getBigtableTableId();

    void setBigtableTableId(String bigtableTableId);
  }
}

Écrire des données dans Bigtable

Pour écrire dans une table Bigtable, vous devez appliquer (apply) une opération CloudBigtableIO.writeToTable. Cette opération doit être effectuée sur une collection PCollection d'objets HBase Mutation, qui peut inclure des objets Put et Delete.

Remarque : Les opérations Append et Increment ne conviennent pas aux modèles de programmation par lot pouvant faire l'objet de plusieurs tentatives d'exécution, y compris Hadoop et Cloud Dataflow, et ne sont donc pas compatibles avec CloudBigtableIO.writeToTable. Les lots Dataflow, ou un groupe d'entrées, peuvent échouer même si certaines des entrées ont été traitées. Dans ce cas, l'ensemble du paquet sera représenté et les opérations Append et Increment précédemment effectuées seront effectuées une seconde fois, entraînant des données incorrectes.

La table Bigtable doit déjà exister et les familles de colonnes appropriées doivent être définies. Le connecteur Dataflow ne crée pas de tables ni de familles de colonnes à la volée. Vous avez la possibilité d'utiliser l'outil de ligne de commande cbt afin de créer une table et de configurer des familles de colonnes, ou vous pouvez le faire de manière automatisée.

Avant d'écrire dans Bigtable, vous devez créer votre pipeline Cloud Dataflow afin que les objets "put" et "delete" puissent être sérialisés sur le réseau :

HelloWorldWrite.java

Afficher sur GitHub

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

En général, vous devez effectuer une transformation, telle que ParDo, pour mettre en forme vos données de sortie en une collection d'objets HBase Put ou Delete. L'exemple suivant illustre une transformation DoFn simple qui prend la valeur actuelle et l'utilise comme clé de ligne pour un objet Put : Vous pouvez ensuite écrire les objets Put dans Bigtable :

HelloWorldWrite.java

Afficher sur GitHub

p.apply(Create.of("phone#4c410523#20190501", "phone#4c410523#20190502"))
    .apply(
        ParDo.of(
            new DoFn<String, Mutation>() {
              @ProcessElement
              public void processElement(@Element String rowkey, OutputReceiver<Mutation> out) {
                long timestamp = System.currentTimeMillis();
                Put row = new Put(Bytes.toBytes(rowkey));

                row.addColumn(
                    Bytes.toBytes("stats_summary"),
                    Bytes.toBytes("os_build"),
                    timestamp,
                    Bytes.toBytes("android"));
                out.output(row);
              }
            }))
    .apply(CloudBigtableIO.writeToTable(bigtableTableConfig));

Exemple d'écriture complet.

HelloWorldWrite.java

Afficher sur GitHub

import com.google.cloud.bigtable.beam.CloudBigtableIO;
import com.google.cloud.bigtable.beam.CloudBigtableTableConfiguration;
import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.options.Default;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;

public class HelloWorldWrite {
  public static void main(String[] args) {
    BigtableOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(BigtableOptions.class);
    Pipeline p = Pipeline.create(options);

    CloudBigtableTableConfiguration bigtableTableConfig =
        new CloudBigtableTableConfiguration.Builder()
            .withProjectId(options.getBigtableProjectId())
            .withInstanceId(options.getBigtableInstanceId())
            .withTableId(options.getBigtableTableId())
            .build();

    p.apply(Create.of("phone#4c410523#20190501", "phone#4c410523#20190502"))
        .apply(
            ParDo.of(
                new DoFn<String, Mutation>() {
                  @ProcessElement
                  public void processElement(@Element String rowkey, OutputReceiver<Mutation> out) {
                    long timestamp = System.currentTimeMillis();
                    Put row = new Put(Bytes.toBytes(rowkey));

                    row.addColumn(
                        Bytes.toBytes("stats_summary"),
                        Bytes.toBytes("os_build"),
                        timestamp,
                        Bytes.toBytes("android"));
                    out.output(row);
                  }
                }))
        .apply(CloudBigtableIO.writeToTable(bigtableTableConfig));

    p.run().waitUntilFinish();
  }

  public interface BigtableOptions extends DataflowPipelineOptions {
    @Description("The Bigtable project ID, this can be different than your Dataflow project")
    @Default.String("bigtable-project")
    String getBigtableProjectId();

    void setBigtableProjectId(String bigtableProjectId);

    @Description("The Bigtable instance ID")
    @Default.String("bigtable-instance")
    String getBigtableInstanceId();

    void setBigtableInstanceId(String bigtableInstanceId);

    @Description("The Bigtable table ID in the instance.")
    @Default.String("bigtable-table")
    String getBigtableTableId();

    void setBigtableTableId(String bigtableTableId);
  }
}