Procesa un flujo de cambios de Bigtable

En este instructivo, se muestra cómo implementar una canalización de datos en Dataflow para obtener una transmisión en tiempo real de los cambios de la base de datos provenientes del flujo de cambios de una tabla de Bigtable. El resultado de la canalización se escribe en una serie de archivos en Cloud Storage.

Se proporciona un conjunto de datos de ejemplo para una aplicación de escucha de música. En este instructivo, se hace un seguimiento de las canciones que se escuchan y, luego, se clasifican las cinco mejores durante un período.

Este instructivo está dirigido a usuarios técnicos que estén familiarizados con la escritura de código y la implementación de canalizaciones de datos en Google Cloud.

Objetivos

En este instructivo, se muestra cómo realizar lo siguiente:

Crea una tabla de Bigtable con un flujo de cambios habilitado.
Implementa una canalización en Dataflow que transforme y genere el flujo de cambios.
Visualiza los resultados de tu canalización de datos.

Costos

En este documento, usarás los siguientes componentes facturables de Google Cloud:

Para generar una estimación de costos en función del uso previsto, usa la calculadora de precios.

Es posible que los usuarios de Google Cloud nuevos cumplan con los requisitos para acceder a una prueba gratuita.

Cuando completes las tareas que se describen en este documento, podrás borrar los recursos que creaste para evitar que se te siga facturando. Para obtener más información, consulta Realiza una limpieza.

Antes de comenzar

Accede a tu Google Cloud cuenta de. Si eres nuevo en Google Cloud, crea una cuenta para evaluar el rendimiento de nuestros productos en situaciones reales. Los clientes nuevos también obtienen $300 en créditos gratuitos para ejecutar, probar y, además, implementar cargas de trabajo.

Instala la Google Cloud CLI. Después de la instalación, inicializa Google Cloud CLI con el siguiente comando:

gcloud init

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

Crea o selecciona un Google Cloud proyecto.

Roles necesarios para seleccionar o crear un proyecto

Seleccionar un proyecto: Para seleccionar un proyecto, no se requiere un rol de IAM específico. Puedes seleccionar cualquier proyecto en el que se te haya otorgado un rol.
Crear un proyecto: Para crear un proyecto, necesitas el rol de creador de proyectos (roles/resourcemanager.projectCreator), que contiene el resourcemanager.projects.create permiso. Obtén información para otorgar roles.

Crea un proyecto de: Google Cloud
```
gcloud projects create PROJECT_ID
```
Reemplaza PROJECT_ID por un nombre para el Google Cloud proyecto de que estás creando.
Selecciona el Google Cloud proyecto de que creaste:
```
gcloud config set project PROJECT_ID
```
Reemplaza PROJECT_ID por el nombre de tu Google Cloud proyecto de.

Verifica que la facturación esté habilitada para tu Google Cloud proyecto.

Habilita las APIs de Dataflow, Cloud Bigtable, API de Cloud Bigtable Admin y Cloud Storage:

Roles necesarios para habilitar las APIs

Para habilitar las APIs, necesitas el rol de IAM de administrador de Service Usage (roles/serviceusage.serviceUsageAdmin), que contiene el serviceusage.services.enable permiso. Obtén información para otorgar roles.

gcloud services enable dataflow.googleapis.com bigtable.googleapis.com bigtableadmin.googleapis.com storage.googleapis.com

Instala la Google Cloud CLI. Después de la instalación, inicializa Google Cloud CLI con el siguiente comando:

gcloud init

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

Crea o selecciona un Google Cloud proyecto.

Roles necesarios para seleccionar o crear un proyecto

Seleccionar un proyecto: Para seleccionar un proyecto, no se requiere un rol de IAM específico. Puedes seleccionar cualquier proyecto en el que se te haya otorgado un rol.
Crear un proyecto: Para crear un proyecto, necesitas el rol de creador de proyectos (roles/resourcemanager.projectCreator), que contiene el resourcemanager.projects.create permiso. Obtén información para otorgar roles.

Crea un proyecto de: Google Cloud
```
gcloud projects create PROJECT_ID
```
Reemplaza PROJECT_ID por un nombre para el Google Cloud proyecto de que estás creando.
Selecciona el Google Cloud proyecto de que creaste:
```
gcloud config set project PROJECT_ID
```
Reemplaza PROJECT_ID por el nombre de tu Google Cloud proyecto de.

Verifica que la facturación esté habilitada para tu Google Cloud proyecto.

Habilita las APIs de Dataflow, Cloud Bigtable, API de Cloud Bigtable Admin y Cloud Storage:

Roles necesarios para habilitar las APIs

gcloud services enable dataflow.googleapis.com bigtable.googleapis.com bigtableadmin.googleapis.com storage.googleapis.com

Actualiza e instala la cbt CLI .

gcloud components update
gcloud components install cbt

Prepare el entorno

Obtén el código

Clona el repositorio que contiene el código de muestra. Si ya descargaste este repositorio, haz un pull para obtener la versión más reciente.

git clone https://github.com/GoogleCloudPlatform/java-docs-samples.git
cd java-docs-samples/bigtable/beam/change-streams

Crea un bucket

Crea un bucket de Cloud Storage:

gcloud storage buckets create gs://BUCKET_NAME

Reemplaza BUCKET_NAME por un nombre de bucket que cumpla con los requisitos de nombre del bucket.

Crear una instancia de Bigtable.

Puedes usar una instancia existente para este instructivo o crear una instancia con las configuraciones predeterminadas en una región cercana.

Crea una tabla

La aplicación de ejemplo hace un seguimiento de las canciones que escuchan los usuarios y almacena los eventos de escucha en Bigtable. Crea una tabla con un flujo de cambios habilitado que tenga una familia de columnas (cf) y una columna (song) y que use IDs de usuario para las claves de fila.

Crea la tabla.

gcloud bigtable instances tables create song-rank \
--column-families=cf --change-stream-retention-period=7d \
--instance=BIGTABLE_INSTANCE_ID --project=PROJECT_ID

Reemplaza lo siguiente:

PROJECT_ID: Es el ID del proyecto que estás usando.
BIGTABLE_INSTANCE_ID: Es el ID de la instancia que contendrá la tabla nueva.

Comienza la canalización

Esta canalización transforma el flujo de cambios de la siguiente manera:

Lee el flujo de cambios.
Obtiene el nombre de la canción.
Agrupa los eventos de escucha de canciones en ventanas de N segundos.
Cuenta las cinco canciones principales.
Genera los resultados.

Ejecutar la canalización

mvn compile exec:java -Dexec.mainClass=SongRank \
"-Dexec.args=--project=PROJECT_ID --bigtableProjectId=PROJECT_ID \
--bigtableInstanceId=BIGTABLE_INSTANCE_ID --bigtableTableId=song-rank \
--outputLocation=gs://BUCKET_NAME/ \
--runner=dataflow --region=BIGTABLE_REGION --experiments=use_runner_v2"

Reemplaza BIGTABLE_REGION por el ID de la región en la que se encuentra tu instancia de Bigtable, como us-east5.

Comprende la canalización

Los siguientes fragmentos de código de la canalización pueden ayudarte a comprender el código que estás ejecutando.

Lectura del flujo de cambios

El código de este ejemplo configura el flujo de origen con los parámetros de la instancia y la tabla de Bigtable específicas.

p.apply(
        "Stream from Bigtable",
        BigtableIO.readChangeStream()
            .withProjectId(options.getBigtableProjectId())
            .withInstanceId(options.getBigtableInstanceId())
            .withTableId(options.getBigtableTableId())
            .withAppProfileId(options.getBigtableAppProfile())

    )

Obtén el nombre de la canción

Cuando se escucha una canción, el nombre de la canción se escribe en la familia de columnas cf y el calificador de columna song, por lo que el código extrae el valor de la mutación del flujo de cambios y lo envía al siguiente paso de la canalización.

private static class ExtractSongName extends DoFn<KV<ByteString, ChangeStreamMutation>, String> {

  @DoFn.ProcessElement
  public void processElement(ProcessContext c) {

    for (Entry e : Objects.requireNonNull(Objects.requireNonNull(c.element()).getValue())
        .getEntries()) {
      if (e instanceof SetCell) {
        SetCell setCell = (SetCell) e;
        if ("cf".equals(setCell.getFamilyName())
            && "song".equals(setCell.getQualifier().toStringUtf8())) {
          c.output(setCell.getValue().toStringUtf8());
        }
      }
    }
  }
}

Cuenta las cinco canciones principales

Puedes usar las funciones integradas de Beam Count y Top.of para obtener las cinco canciones principales en la ventana actual.

.apply(Count.perElement())
.apply("Top songs", Top.of(5, new SongComparator()).withoutDefaults())

Genera los resultados

Esta canalización escribe los resultados en la salida estándar y en los archivos. Para los archivos, muestra las escrituras en ventanas en grupos de 10 elementos o segmentos de un minuto.

.apply("Print", ParDo.of(new PrintFn()))
.apply(
    "Collect at least 10 elements or 1 minute of elements",
    Window.<String>into(new GlobalWindows())
        .triggering(
            Repeatedly.forever(
                AfterFirst.of(
                    AfterPane.elementCountAtLeast(10),
                    AfterProcessingTime
                        .pastFirstElementInPane()
                        .plusDelayOf(Duration.standardMinutes(1)
                        )
                )
            ))
        .discardingFiredPanes())
.apply(
    "Output top songs",
    TextIO.write()
        .to(options.getOutputLocation() + "song-charts/")
        .withSuffix(".txt")
        .withNumShards(1)
        .withWindowedWrites()
);

Visualiza la canalización

En la Google Cloud consola de, ve a la página Dataflow.

Ir a Dataflow
Haz clic en el trabajo con un nombre que comience con song-rank.
En la parte inferior de la pantalla, haz clic en Mostrar para abrir el panel de registros.
Haz clic en Registros de trabajadores para supervisar los registros de salida del flujo de cambios.

Nota: El registro puede tardar unos minutos en aparecer mientras se inicializa el trabajo.

Operaciones de escritura por transmisión

Usa la CLI de cbt para escribir una cantidad de escuchas de canciones para varios usuarios en la tabla song-rank. Esto está diseñado para escribir durante unos minutos para simular la transmisión de escuchas de canciones a lo largo del tiempo.

cbt -instance=BIGTABLE_INSTANCE_ID -project=PROJECT_ID import \
song-rank song-rank-data.csv  column-family=cf batch-size=1

Revise el resultado.

Lee el resultado en Cloud Storage para ver las canciones más populares.

gcloud storage cat gs://BUCKET_NAME/song-charts/GlobalWindow-pane-0-00000-of-00001.txt

Resultado de ejemplo:

2023-07-06T19:53:38.232Z [KV{The Wheels on the Bus, 199}, KV{Twinkle, Twinkle, Little Star, 199}, KV{Ode to Joy , 192}, KV{Row, Row, Row Your Boat, 186}, KV{Take Me Out to the Ball Game, 182}]
2023-07-06T19:53:49.536Z [KV{Old MacDonald Had a Farm, 20}, KV{Take Me Out to the Ball Game, 18}, KV{Für Elise, 17}, KV{Ode to Joy , 15}, KV{Mary Had a Little Lamb, 12}]
2023-07-06T19:53:50.425Z [KV{Twinkle, Twinkle, Little Star, 20}, KV{The Wheels on the Bus, 17}, KV{Row, Row, Row Your Boat, 13}, KV{Happy Birthday to You, 12}, KV{Over the Rainbow, 9}]

Limpia

Para evitar que se apliquen cargos a tu cuenta de Google Cloud por los recursos usados en este instructivo, borra el proyecto que contiene los recursos o conserva el proyecto y borra los recursos individuales.

Borra el proyecto

Precaución: Borrar un proyecto tiene estas consecuencias:

Se borra todo su contenido. Si usaste un proyecto existente para las tareas de este documento, cuando lo borres, también se borrará cualquier otro trabajo que hayas realizado en el proyecto.
Se pierden los IDs personalizados de los proyectos. Cuando creaste este proyecto, es posible que hayas creado un ID personalizado del proyecto que desees utilizar en el futuro. Para conservar las URLs que usan el ID del proyecto, como una URL appspot.com, borra los recursos seleccionados dentro del proyecto, en lugar de todo el proyecto.

Si planeas explorar varios instructivos, arquitecturas o guías de inicio rápido, reutilizar proyectos puede ser útil para no exceder los límites de cuota de los proyectos.

Borra un Google Cloud proyecto de:

gcloud projects delete PROJECT_ID

Borra los recursos individuales

Borra el bucket y los archivos.

gcloud storage rm --recursive gs://BUCKET_NAME/

Inhabilita el flujo de cambios en la tabla.

gcloud bigtable instances tables update song-rank --instance=BIGTABLE_INSTANCE_ID \
--clear-change-stream-retention-period

Borra la tabla song-rank.

cbt -instance=BIGTABLE_INSTANCE_ID -project=PROJECT_ID deletetable song-rank

Detén la canalización de flujo de cambios.
1. Enumera los trabajos para obtener el ID del trabajo.
```
gcloud dataflow jobs list --region=BIGTABLE_REGION
```
2. Cancela el trabajo.
```
gcloud dataflow jobs cancel JOB_ID --region=BIGTABLE_REGION
```
  Reemplaza JOB_ID por el ID de trabajo que se muestra después del comando anterior.

Procesa un flujo de cambios de Bigtable Organiza tus páginas con colecciones Guarda y categoriza el contenido según tus preferencias.

Objetivos

Costos

Antes de comenzar

Prepare el entorno

Obtén el código

Crea un bucket

Crear una instancia de Bigtable.

Crea una tabla

Comienza la canalización

Comprende la canalización

Lectura del flujo de cambios

Obtén el nombre de la canción

Cuenta las cinco canciones principales

Genera los resultados

Visualiza la canalización

Operaciones de escritura por transmisión

Revise el resultado.

Limpia

Borra el proyecto

Borra los recursos individuales

¿Qué sigue?

Procesa un flujo de cambios de Bigtable