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 que provienen 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, harás un seguimiento de las canciones que se escuchan y, luego, ordenarás las cinco principales en un período.
Este instructivo está dirigido a usuarios técnicos familiarizados con la escritura de código y para implementar 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.
- Consulta 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.
Cuando finalices las tareas que se describen en este documento, puedes borrar los recursos que creaste para evitar que continúe la facturación. Para obtener más información, consulta Cómo realizar una limpieza.
Antes de comenzar
- Accede a tu cuenta de Google Cloud. 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.
- Install the Google Cloud CLI.
-
To initialize the gcloud CLI, run the following command:
gcloud init
-
Create or select a Google Cloud project.
-
Create a Google Cloud project:
gcloud projects create PROJECT_ID
Replace
PROJECT_ID
with a name for the Google Cloud project you are creating. -
Select the Google Cloud project that you created:
gcloud config set project PROJECT_ID
Replace
PROJECT_ID
with your Google Cloud project name.
-
-
Asegúrate de que la facturación esté habilitada para tu proyecto de Google Cloud.
-
Enable the Dataflow, Cloud Bigtable API, Cloud Bigtable Admin API, and Cloud Storage APIs:
gcloud services enable dataflow.googleapis.com
bigtable.googleapis.com bigtableadmin.googleapis.com storage.googleapis.com - Install the Google Cloud CLI.
-
To initialize the gcloud CLI, run the following command:
gcloud init
-
Create or select a Google Cloud project.
-
Create a Google Cloud project:
gcloud projects create PROJECT_ID
Replace
PROJECT_ID
with a name for the Google Cloud project you are creating. -
Select the Google Cloud project that you created:
gcloud config set project PROJECT_ID
Replace
PROJECT_ID
with your Google Cloud project name.
-
-
Asegúrate de que la facturación esté habilitada para tu proyecto de Google Cloud.
-
Enable the Dataflow, Cloud Bigtable API, Cloud Bigtable Admin API, and Cloud Storage APIs:
gcloud services enable dataflow.googleapis.com
bigtable.googleapis.com bigtableadmin.googleapis.com storage.googleapis.com - Actualiza e instala el
CLI de
cbt
de Google Cloud.gcloud components update gcloud components install cbt
Prepare el entorno
Cómo obtener el código
Clona el repositorio que contiene el código de muestra. Si ya descargaste este repositorio, extrae 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
gcloud storage buckets create gs://BUCKET_NAME
BUCKET_NAME
with a bucket name
that meets the bucket naming requirements.
Crear una instancia de Bigtable.
Puedes usar una instancia existente para este instructivo o crear una instancia con la configuración predeterminada en una región cercana.
Crea una tabla
La aplicación de ejemplo realiza un seguimiento de las canciones que escuchan los usuarios y almacena los eventos de reproducción en Bigtable. Crea una tabla con un flujo de cambios habilitado que tenga una familia de columnas (cf) y una columna (canción) 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 usas.
- 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 reproducción de la canción en ventanas de N segundos.
- Cuenta las cinco canciones más populares
- Muestra 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.
Cómo leer el 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.
Obteniendo 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.
Cómo contar las cinco canciones principales
Puedes usar las funciones integradas de Beam Count
y Top.of
para obtener las cinco
canciones en la ventana actual.
Cómo generar los resultados
Esta canalización escribe los resultados en la salida estándar y en los archivos. En el caso de los archivos, agrupa las operaciones de escritura en grupos de 10 elementos o segmentos de un minuto.
Visualiza la canalización
En la consola de Google Cloud, ve a la página Dataflow.
Haz clic en el trabajo que tenga 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 trabajador para supervisar los registros de salida del flujo de cambios.
Operaciones de escritura de transmisión
Usa la CLI de cbt
para escribir una cantidad de reproducciones de canciones de varios usuarios en la tabla song-rank
. Se diseñó para escribir en unos minutos para simular
la canción que se está reproduciendo 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
Salida 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
Borra un proyecto de Google Cloud:
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 flujos de cambios.
Enumera los trabajos para obtener el ID de trabajo.
gcloud dataflow jobs list --region=BIGTABLE_REGION
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.