Dataflow es un servicio completamente administrado para transformar y enriquecer datos en modos de transmisión (tiempo real) y por lotes con la misma confiabilidad y expresividad. Proporciona un entorno de desarrollo de canalización simplificado con el SDK de Apache Beam, que tiene un conjunto amplio de primitivas de análisis de sesiones y sistemas de ventanas, además de un ecosistema de conectores fuente y receptores. En esta guía de inicio rápido, se muestra cómo usar Dataflow para realizar las siguientes acciones:
- Leer mensajes publicados en un tema de Pub/Sub
- Mostrar mensajes en ventanas, o agruparlos, por marca de tiempo
- Escribir mensajes a Cloud Storage
En esta guía de inicio rápido, se explica el uso de Dataflow en Java y Python. SQL también es compatible.
Si tu intención no es realizar un procesamiento de datos personalizado, puedes comenzar a usar las plantillas de Dataflow basadas en IU.
Antes de comenzar
- Sigue las instrucciones para instalar e inicializar el SDK de Cloud.
- Habilita la facturación en tu proyecto.
Para completar esta guía de inicio rápido, debes habilitar las siguientes API: Compute Engine, Google Cloud's operations suite, Cloud Storage, JSON de Cloud Storage, Pub/Sub, Cloud Scheduler, Resource Manager y App Engine.
Pueden pasar unos minutos hasta que las API aparezcan en Console.
Para crear una clave de cuenta de servicio, haz lo siguiente:
Cree una clave de cuenta de servicio
- En la lista Cuenta de servicio, selecciona Cuenta de servicio nueva.
- Ingresa un nombre en el campo Nombre de cuenta de servicio.
- En la lista Función, selecciona Proyecto > Propietario.
- Haz clic en Crear.
La clave se envía a la carpeta de descargas predeterminada del navegador.
Establece la variable de entorno
GOOGLE_APPLICATION_CREDENTIALS
para que apunte a la clave de la cuenta de servicio.export GOOGLE_APPLICATION_CREDENTIALS=path/to/my/credentials.json
Crea variables para tu bucket, tu proyecto y la región. Los nombres de depósitos de Cloud Storage deben ser únicos a nivel global. Selecciona una región de Dataflow cercana a donde ejecutas los comandos de esta guía de inicio rápido.
BUCKET_NAME=your-bucket-name PROJECT_ID=$(gcloud config get-value project) TOPIC_ID=your-topic-id REGION=dataflow-region
Crea un bucket de Cloud Storage que sea propiedad de este proyecto:
gsutil mb gs://$BUCKET_NAME
Crea un tema de Pub/Sub en este proyecto:
gcloud pubsub topics create $TOPIC_ID
Crea un trabajo de Cloud Scheduler en este proyecto. El trabajo publica un mensaje en un tema de Cloud Pub/Sub con intervalos de un minuto.
Si una app de App Engine no existe en el proyecto, crea una con el siguiente comando.
gcloud scheduler jobs create pubsub publisher-job --schedule="* * * * *" \ --topic=$TOPIC_ID --message-body="Hello!"
Inicia el trabajo.
gcloud scheduler jobs run publisher-job
Usa el siguiente comando para clonar el repositorio de la guía de inicio rápido y navega al directorio de código de muestra:
Java
git clone https://github.com/GoogleCloudPlatform/java-docs-samples.git cd java-docs-samples/pubsub/streaming-analytics
Python
git clone https://github.com/GoogleCloudPlatform/python-docs-samples.git cd python-docs-samples/pubsub/streaming-analytics pip install -r requirements.txt # Install Apache Beam dependencies
Transmite mensajes desde Pub/Sub a Cloud Storage
Muestra de código
En este código de muestra, se usa Dataflow para realizar las siguientes acciones:
- Leer mensajes de Pub/Sub
- Mostrar mensajes en ventanas, o agruparlos, en intervalos de tamaño fijo con marcas de tiempo públicas.
Escribir los mensajes en cada ventana en archivos en Cloud Storage
Java
Python
Comienza la canalización
Para iniciar la canalización, ejecuta el siguiente comando:
Java
mvn compile exec:java \ -Dexec.mainClass=com.examples.pubsub.streaming.PubSubToGcs \ -Dexec.cleanupDaemonThreads=false \ -Dexec.args=" \ --project=$PROJECT_ID \ --region=$REGION \ --inputTopic=projects/$PROJECT_ID/topics/$TOPIC_ID \ --output=gs://$BUCKET_NAME/samples/output \ --runner=DataflowRunner \ --windowSize=2"
Python
python PubSubToGCS.py \ --project=$PROJECT_ID \ --region=$REGION \ --input_topic=projects/$PROJECT_ID/topics/$TOPIC_ID \ --output_path=gs://$BUCKET_NAME/samples/output \ --runner=DataflowRunner \ --window_size=2 \ --num_shards=2 \ --temp_location=gs://$BUCKET_NAME/temp
El comando anterior se ejecuta de manera local y, luego, inicia un trabajo de Dataflow que se ejecuta en la nube. Cuando el comando muestre JOB_MESSAGE_DETAILED: Workers
have started successfully
, sal del programa local con Ctrl+C
.
Observa el progreso del trabajo y la canalización
Puedes observar el progreso del trabajo en la consola de Dataflow.
Abre la vista de detalles de trabajos para ver lo siguiente:
- Estructura del trabajo
- Registros del trabajo
- Métricas de etapas
Puede que debas esperar unos minutos para ver los archivos de salida en Cloud Storage.
También puedes usar la línea de comandos que se muestra a continuación para verificar qué archivos se escribieron.
gsutil ls gs://${BUCKET_NAME}/samples/
El resultado debe tener el siguiente aspecto:
Java
gs://{$BUCKET_NAME}/samples/output-22:30-22:32-0-of-1 gs://{$BUCKET_NAME}/samples/output-22:32-22:34-0-of-1 gs://{$BUCKET_NAME}/samples/output-22:34-22:36-0-of-1 gs://{$BUCKET_NAME}/samples/output-22:36-22:38-0-of-1
Python
gs://{$BUCKET_NAME}/samples/output-22:30-22:32-0 gs://{$BUCKET_NAME}/samples/output-22:30-22:32-1 gs://{$BUCKET_NAME}/samples/output-22:32-22:34-0 gs://{$BUCKET_NAME}/samples/output-22:32-22:34-1
Limpieza
Borra el trabajo de Cloud Scheduler.
gcloud scheduler jobs delete publisher-job
En la consola de Dataflow, detén el trabajo. Cancela la canalización sin desviarla.
Borra el tema.
gcloud pubsub topics delete $TOPIC_ID
Borra los archivos que se crearon con la canalización.
gsutil -m rm -rf "gs://${BUCKET_NAME}/samples/output*" gsutil -m rm -rf "gs://${BUCKET_NAME}/temp/*"
Quita el bucket de Cloud Storage.
gsutil rb gs://${BUCKET_NAME}
¿Qué sigue?
Si deseas mostrar los mensajes de Pub/Sub en una marca de tiempo personalizada, puedes especificar la marca de tiempo como un atributo en el mensaje de Pub/Sub y, luego, usar la marca de tiempo personalizada con el comando
withTimestampAttribute
de PubsubIO.Observa las plantillas de código abierto de Dataflow diseñadas para la transmisión de Google.
Obtén más información sobre cómo Dataflow se integra com Pub/Sub.
Mira este instructivo que lee desde Pub/Sub y escribe en BigQuery con las plantillas de Dataflow Flex.
Para obtener más información sobre el sistema de ventanas, consulta el ejemplo en la página sobre canalización de videojuegos para dispositivos móviles de Apache Beam.