Se usó la API de Cloud Translation para traducir esta página.

Transmite cambios con Dataflow

El conector de Bigtable Beam te permite usar Dataflow para leer registros de cambios de datos de Bigtable sin necesidad de hacer un seguimiento de los cambios de partición ni procesarlos en tu código, ya que el conector controla esa lógica por ti.

En este documento, se describe cómo configurar y usar el conector de Bigtable Beam para leer un flujo de cambios con una canalización de Dataflow. Antes de leer este documento, debes leer la Descripción general de los flujos de cambios y familiarizarte con Dataflow.

Alternativas para compilar tu propia canalización

Si no quieres compilar tu propia canalización de Dataflow, puedes usar una de las siguientes opciones.

Puedes usar una plantilla de Dataflow proporcionada por Google.

También puedes usar las muestras de código del instructivo o la guía de inicio rápido de Bigtable como punto de partida para tu código.

Asegúrate de que el código que generes use la versión 26.14.0 o posterior de google cloud libraries-bom.

Detalles del conector

El método del conector de Bigtable Beam, BigtableIO.readChangeStream, te permite leer un flujo de registros de cambios de datos (ChangeStreamMutation) que puedes procesar. El conector de Bigtable Beam es un componente del repositorio de GitHub de Apache Beam. Para obtener una descripción del código del conector, consulta los comentarios en BigtableIO.java.

Debes usar el conector con la versión 2.48.0 o posterior de Beam. Consulta Compatibilidad con el entorno de ejecución de Apache Beam para asegurarte de que estás usando una versión compatible de Java. Luego, puedes implementar una canalización que use el conector para Dataflow, que se encarga del aprovisionamiento y la administración de recursos, y ayuda con la escalabilidad y la confiabilidad del procesamiento de datos de transmisión.

Para obtener más información sobre el modelo de programación de Apache Beam, consulta la documentación de Beam.

Cómo agrupar datos sin horas de eventos

Los registros de cambios de datos transmitidos con el conector de Bigtable Beam no son compatibles con las funciones de Dataflow que dependen de las marcas de tiempo de eventos.

Como se explica en Replicación y marcas de agua, es posible que una marca de agua baja no avance si la replicación de la partición no se puso al día con el resto de la instancia. Cuando una marca de agua baja deja de avanzar, puede provocar que el flujo de cambios se detenga.

Para evitar que la transmisión se detenga, el conector de Bigtable Beam genera todos los datos con una marca de tiempo de salida de cero. La marca de tiempo cero hace que Dataflow considere que todos los registros de cambio de datos son datos atrasados. Como resultado, las funciones de Dataflow que dependen de las marcas de tiempo de eventos no son compatibles con los flujos de cambios de Bigtable. Específicamente, no puedes usar funciones de ventana, activadores de tiempo de evento ni temporizadores de tiempo de evento.

En cambio, puedes usar GlobalWindows con activadores que no sean de tiempo de evento para agrupar estos datos tardíos en paneles, como se muestra en el ejemplo del instructivo. Para obtener detalles sobre los activadores y los paneles, consulta Activadores en la guía de programación de Beam.

Ajuste de escala automático

El conector admite el ajuste de escala automático de Dataflow, que está habilitado de forma predeterminada cuando se usa Runner v2 (obligatorio). El algoritmo de ajuste de escala automático de Dataflow tiene en cuenta la acumulación estimada de transmisiones de cambios, que se puede supervisar en la página Supervisión de Dataflow en la sección Backlog. Usa la marca --maxNumWorkers cuando implementes un trabajo para limitar la cantidad de trabajadores.

Para escalar tu canalización de forma manual en lugar de usar el ajuste de escala automático, consulta Cómo escalar una canalización de transmisión de forma manual.

Limitaciones

Ten en cuenta las siguientes limitaciones antes de usar el conector de Bigtable Beam con Dataflow.

Dataflow Runner V2

El conector solo se puede ejecutar con Dataflow Runner v2. Para habilitar esta opción, especifica --experiments=use_runner_v2 en los argumentos de la línea de comandos. Si ejecutas con Runner v1, la canalización fallará con la siguiente excepción:

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

Instantáneas

El conector no admite instantáneas de Dataflow.

Duplicados

El conector de Bigtable Beam transmite los cambios de cada clave de fila y cada clúster en orden de marca de tiempo de confirmación, pero, como a veces se reinicia desde momentos anteriores de la transmisión, puede producir duplicados.

Se reinicia la canalización

Si una canalización de Dataflow se detuvo durante mucho tiempo, los registros de cambio de datos pueden quedar rezagados con respecto al límite de retención. Cuando se reanuda la canalización, Bigtable la falla para que puedas iniciar una nueva con una nueva hora de inicio de la solicitud que se encuentre dentro del período de retención. Bigtable hace esto, en lugar de adelantar silenciosamente la hora de la solicitud de la canalización original, para evitar la eliminación no intencional de registros de cambios de datos con marcas de tiempo que quedan fuera del período de retención especificado.

Antes de comenzar

Antes de usar el conector, completa los siguientes requisitos previos.

Configura la autenticación

Para usar las muestras de Java de esta página en un entorno de desarrollo local, instala e inicializa gcloud CLI y, luego, configura las credenciales predeterminadas de la aplicación con tus credenciales de usuario.

Instala Google Cloud CLI.

Si usas un proveedor de identidad externo (IdP), primero debes acceder a gcloud CLI con tu identidad federada.

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.

If an authentication error is returned, and you are using an external identity provider (IdP), confirm that you have signed in to the gcloud CLI with your federated identity.

Para obtener más información, consulta Set up authentication for a local development environment.

Si quieres obtener información sobre cómo configurar la autenticación para un entorno de producción, consulta Set up Application Default Credentials for code running on Google Cloud.

Habilita un flujo de cambios

Debes habilitar un flujo de cambios en una tabla antes de poder leerla. También puedes crear una tabla nueva con las flujos de cambios habilitadas.

Tabla de metadatos de cambio de transmisión

Cuando transmites cambios con Dataflow, el conector de Bigtable Beam crea una tabla de metadatos que, de forma predeterminada, se denomina __change_stream_md_table. La tabla de metadatos del flujo de cambios administra el estado operativo del conector y almacena metadatos sobre los registros de cambios en los datos.

De forma predeterminada, el conector crea la tabla en la misma instancia que la tabla que se transmite. Para asegurarte de que la tabla funcione correctamente, el perfil de la app para la tabla de metadatos debe usar el enrutamiento de un solo clúster y tener habilitadas las transacciones de fila única.

Para obtener más información sobre la transmisión de cambios desde Bigtable con el conector de Bigtable Beam, consulta la documentación de BigtableIO.

Roles obligatorios

Para obtener los permisos que necesitas para leer un flujo de cambios de Bigtable con Dataflow, pídele a tu administrador que te otorgue los siguientes roles de IAM.

Para leer los cambios de Bigtable, necesitas este rol:

Administrador de Bigtable (roles/bigtable.admin) en la instancia de Bigtable que contiene la tabla de la que planeas transmitir los cambios

Para ejecutar el trabajo de Dataflow, necesitas estos roles:

Desarrollador de Dataflow (roles/dataflow.developer) en el proyecto que contiene tus recursos de Cloud
Trabajador de Dataflow (roles/dataflow.worker) en el proyecto que contiene tus recursos de Cloud
Administrador de objetos de Storage (roles/storage.objectAdmin) en los buckets de Cloud Storage que planeas usar

Para obtener más información sobre cómo otorgar roles, consulta Administra el acceso.

También puedes obtener los permisos necesarios mediante roles personalizados o cualquier otro rol predefinido.

Agrega el conector de Bigtable Beam como dependencia

Agrega código similar a la siguiente dependencia a tu archivo pom.xml de Maven. La versión debe ser 2.48.0 o posterior.

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

Leer el flujo de cambios

Para crear una canalización de Dataflow que lea tus registros de cambios de datos, configura el conector y, luego, agrega transformaciones y receptores. Luego, usas el conector para leer objetos ChangeStreamMutation en un canal de Beam.

En los ejemplos de código de esta sección, escritos en Java, se muestra cómo compilar una canalización y usarla para convertir pares clave-valor en una cadena. Cada par consta de una clave de fila y un objeto ChangeStreamMutation. La canalización convierte las entradas de cada objeto en una cadena separada por comas.

Compila la canalización

En esta muestra de código Java, se muestra cómo compilar la canalización:

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();

Procesa los registros de cambios en los datos

En este ejemplo, se muestra cómo iterar todas las entradas de un registro de cambio de datos para una fila y llamar a un método de conversión a cadena según el tipo de entrada.

Para obtener una lista de los tipos de entrada que puede contener un registro de cambio de datos, consulta Qué contiene un registro de cambio de datos.

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;
}

En este ejemplo, se convierte una entrada de escritura:

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);
}

En este ejemplo, se convierte una entrada de eliminación de celdas:

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);
}

En este ejemplo, se convierte una entrada de borrado de una familia de columnas:


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);
}

Supervisar

Los siguientes recursos de la consola de Google Cloud te permiten supervisar tus recursos deGoogle Cloud mientras ejecutas una canalización de Dataflow para leer un flujo de cambios de Bigtable:

En particular, verifica las siguientes métricas:

En la página de estadísticas del sistema de Bigtable, verifica las siguientes métricas:
- Datos de uso de CPU por flujos de cambios en la métrica cpu_load_by_app_profile_by_method_by_table. Muestra el impacto del flujo de cambios en el uso de CPU del clúster.
- Utilización del almacenamiento del flujo de cambios (bytes) (change_stream_log_used_bytes).
En la página de supervisión de Dataflow, verifica la actualidad de los datos. Esta métrica muestra la diferencia entre la hora actual y la marca de agua, que es de aproximadamente dos minutos, con picos ocasionales que duran uno o dos minutos más. La actualidad de los datos no indica si los registros de cambios de datos se procesan lentamente. Para garantizar el rendimiento y el estado continuos de tus aplicaciones críticas, supervisa la métrica de actualización de datos de Dataflow y realiza las siguientes acciones:
- Si la métrica de actualidad de los datos es constantemente más alta que el umbral, es posible que tu canalización no tenga suficientes recursos. Te recomendamos que agregues más trabajadores de Dataflow.
- Si los trabajadores de Dataflow están bien aprovisionados, pero la actualidad de los datos ha aumentado o es constantemente alta, comunícate con el equipo de asistencia deGoogle Cloud .
La métrica processing_delay_from_commit_timestamp_MEAN de Dataflow puede indicarte el tiempo de procesamiento promedio de los registros de cambio de datos durante la vida útil del trabajo.

La métrica server/latencies de Bigtable no es útil cuando supervisas una canalización de Dataflow que lee un flujo de cambios de Bigtable, ya que refleja la duración de la solicitud de transmisión, no la latencia del procesamiento de registros de cambios de datos. La latencia alta en un flujo de cambios no significa que las solicitudes se estén procesando lentamente, sino que la conexión estuvo abierta durante ese tiempo.