GSP290

Descripción general

Dataflow es un servicio de Google Cloud que proporciona procesamiento unificado de datos por lotes y de transmisión a gran escala. Se basa en el proyecto Apache Beam, un modelo de código abierto para definir canalizaciones de procesamiento paralelo de datos de transmisión y por lotes. Con uno de los SDK de Apache Beam de código abierto, puedes crear un programa que defina la canalización y, luego, usar Dataflow para ejecutarla.

En este lab, usarás el SDK de Apache Beam para Python para crear y ejecutar una canalización en Dataflow que transfiera datos de Cloud Storage a BigQuery y, luego, los transforme y enriquezca en BigQuery.

Nota: Asegúrate de abrir los archivos de Python y leer los comentarios cuando se te indique para entender lo que hace el código.

Actividades

En este lab, aprenderás a realizar las siguientes tareas:

• Crear y ejecutar canalizaciones de Dataflow (Python) para transferir datos de Cloud Storage a BigQuery
• Crear y ejecutar canalizaciones de Dataflow (Python) para transformar y enriquecer datos en BigQuery
• Crear y ejecutar canalizaciones de Dataflow (Python) para unir datos en BigQuery y escribir los resultados en una tabla nueva

Configuración y requisitos

Antes de hacer clic en el botón Comenzar lab

Lee estas instrucciones. Los labs cuentan con un temporizador que no se puede pausar. El temporizador, que comienza a funcionar cuando haces clic en Comenzar lab, indica por cuánto tiempo tendrás a tu disposición los recursos de Google Cloud.

Este lab práctico te permitirá realizar las actividades correspondientes en un entorno de nube real, no en uno de simulación o demostración. Para ello, se te proporcionan credenciales temporales nuevas que utilizarás para acceder a Google Cloud durante todo el lab.

Para completar este lab, necesitarás lo siguiente:

• Acceso a un navegador de Internet estándar. Se recomienda el navegador Chrome.

Nota: Usa una ventana del navegador privada o de incógnito (opción recomendada) para ejecutar el lab. Así evitarás conflictos entre tu cuenta personal y la cuenta de estudiante, lo que podría generar cargos adicionales en tu cuenta personal.

• Tiempo para completar el lab (recuerda que, una vez que comienzas un lab, no puedes pausarlo).

Nota: Usa solo la cuenta de estudiante para este lab. Si usas otra cuenta de Google Cloud, es posible que se apliquen cargos a esa cuenta.

Cómo iniciar tu lab y acceder a la consola de Google Cloud

1. Haz clic en el botón Comenzar lab. Si debes pagar por el lab, se abrirá un diálogo para que selecciones la forma de pago. A la izquierda, se encuentra el panel Detalles del lab, que tiene estos elementos:

   • El botón para abrir la consola de Google Cloud
   • El tiempo restante
   • Las credenciales temporales que debes usar para el lab
   • Otra información para completar el lab, si es necesaria

2. Haz clic en Abrir la consola de Google Cloud (o haz clic con el botón derecho y selecciona Abrir el vínculo en una ventana de incógnito si ejecutas el navegador Chrome).

   El lab inicia recursos y abre otra pestaña en la que se muestra la página de acceso.

   Sugerencia: Ordena las pestañas en ventanas separadas, una junto a la otra.

   Nota: Si ves el diálogo Elegir una cuenta, haz clic en Usar otra cuenta.

3. De ser necesario, copia el nombre de usuario a continuación y pégalo en el diálogo Acceder.

   {{{user_0.username | "Username"}}}

   También puedes encontrar el nombre de usuario en el panel Detalles del lab.

4. Haz clic en Siguiente.

5. Copia la contraseña que aparece a continuación y pégala en el diálogo Te damos la bienvenida.

   {{{user_0.password | "Password"}}}

   También puedes encontrar la contraseña en el panel Detalles del lab.

6. Haz clic en Siguiente.

   Importante: Debes usar las credenciales que te proporciona el lab. No uses las credenciales de tu cuenta de Google Cloud. Nota: Usar tu propia cuenta de Google Cloud para este lab podría generar cargos adicionales.

7. Haz clic para avanzar por las páginas siguientes:

   • Acepta los Términos y Condiciones.
   • No agregues opciones de recuperación o autenticación de dos factores (esta es una cuenta temporal).
   • No te registres para obtener pruebas gratuitas.

Después de un momento, se abrirá la consola de Google Cloud en esta pestaña.

Nota: Para acceder a los productos y servicios de Google Cloud, haz clic en el menú de navegación o escribe el nombre del servicio o producto en el campo Buscar.

Activa Cloud Shell

Cloud Shell es una máquina virtual que cuenta con herramientas para desarrolladores. Ofrece un directorio principal persistente de 5 GB y se ejecuta en Google Cloud. Cloud Shell proporciona acceso de línea de comandos a tus recursos de Google Cloud.

1. Haz clic en Activar Cloud Shell en la parte superior de la consola de Google Cloud.

2. Haz clic para avanzar por las siguientes ventanas:

   • Continúa en la ventana de información de Cloud Shell.
   • Autoriza a Cloud Shell para que use tus credenciales para realizar llamadas a la API de Google Cloud.

Cuando te conectes, habrás completado la autenticación, y el proyecto estará configurado con tu Project_ID, . El resultado contiene una línea que declara el Project_ID para esta sesión:

Your Cloud Platform project in this session is set to {{{project_0.project_id | "PROJECT_ID"}}}

gcloud es la herramienta de línea de comandos de Google Cloud. Viene preinstalada en Cloud Shell y es compatible con la función de autocompletado con tabulador.

3. Puedes solicitar el nombre de la cuenta activa con este comando (opcional):

gcloud auth list

4. Haz clic en Autorizar.

Resultado:

ACTIVE: * ACCOUNT: {{{user_0.username | "ACCOUNT"}}} To set the active account, run: $ gcloud config set account `ACCOUNT`

5. Puedes solicitar el ID del proyecto con este comando (opcional):

gcloud config list project

Resultado:

[core] project = {{{project_0.project_id | "PROJECT_ID"}}} Nota: Para obtener toda la documentación de gcloud, en Google Cloud, consulta la guía con la descripción general de gcloud CLI.

Tarea 1: Asegúrate de que la API de Dataflow esté habilitada correctamente

Para garantizar el acceso a la API necesaria, reinicia la conexión a la API de Dataflow.

Importante: Incluso si la API está habilitada, sigue los pasos del 1 al 4 que se indican más abajo para inhabilitarla. Luego, vuelve a habilitarla para reiniciar la API correctamente.

1. En la barra de título de la consola de Google Cloud, escribe API de Dataflow en el campo Buscar y, luego, haz clic en API de Dataflow en los resultados.

2. Haz clic en Administrar.

3. Haz clic en Inhabilitar API.

Si se te solicita confirmar, haz clic en Inhabilitar.

4. Haz clic en Habilitar.

Cuando vuelvas a habilitar la API, la página mostrará la opción Inhabilitar.

Haz clic en Revisar mi progreso para verificar la tarea realizada.

Inhabilitar y volver a habilitar la API de Dataflow

Tarea 2: Descarga el código de partida

Descarga los ejemplos de Dataflow para Python para usarlos en este lab.

1. Ejecuta el siguiente comando en Cloud Shell para obtener ejemplos de Dataflow para Python del repositorio professional-services de Google Cloud en GitHub:

gcloud storage cp -r gs://spls/gsp290/dataflow-python-examples .

2. Establece una variable para el ID del proyecto.

export PROJECT={{{ project_0.project_id }}} gcloud config set project $PROJECT

Tarea 3: Crea un bucket de Cloud Storage y copia archivos en él

En Cloud Shell, crea un bucket de Cloud Storage y, luego, copia en él los archivos de los ejemplos de Dataflow para Python.

Crea un bucket de Cloud Storage

• Sin salir de Cloud Shell, usa el comando make bucket para crear un bucket regional en la región de tu proyecto:

gcloud storage buckets create gs://$PROJECT --location={{{ project_0.default_region | REGION }}}

Haz clic en Revisar mi progreso para verificar la tarea realizada.

Crear un bucket de Cloud Storage

Copia archivos en tu bucket

• En Cloud Shell, usa el comando gsutil para copiar archivos en el bucket de Cloud Storage que acabas de crear:

gcloud storage cp gs://spls/gsp290/data_files/usa_names.csv gs://$PROJECT/data_files/ gcloud storage cp gs://spls/gsp290/data_files/head_usa_names.csv gs://$PROJECT/data_files/

Haz clic en Revisar mi progreso para verificar la tarea realizada.

Copiar archivos en tu bucket

Tarea 4: Crea un conjunto de datos de BigQuery

Crea un conjunto de datos en BigQuery, en el que cargarás las tablas.

En Cloud Shell, crea un conjunto de datos llamado lake:

bq mk lake

Haz clic en Revisar mi progreso para verificar la tarea realizada.

Crear un conjunto de datos de BigQuery llamado lake

Tarea 5: Revisa y ejecuta la canalización de transferencia de datos

En esta tarea, revisarás el código de la canalización para ver cómo funciona. Luego, la configurarás y ejecutarás.

Esta canalización transfiere datos de Cloud Storage a la tabla de BigQuery con la fuente TextIO y el destino BigQueryIO. En particular, hace lo siguiente:

• Transferir los archivos desde Cloud Storage
• Filtrar la fila del encabezado en los archivos
• Convertir las líneas leídas en objetos de diccionario
• Guardar las filas como resultados en BigQuery

Revisa el código de Python de la canalización de transferencia de datos

Usa el Editor de código de Cloud Shell para revisar el código de la canalización.

1. En la barra de menú de Cloud Shell, haz clic en Abrir editor.

2. Navega a dataflow_python_examples > dataflow_python_examples y abre el archivo data_ingestion.py.

3. Lee los comentarios del archivo, en los que se explica lo que hace el código.

Este código completa una tabla de BigQuery con los archivos de datos de Cloud Storage.

4. Para volver a Cloud Shell, haz clic en Abrir terminal.

Configura el contenedor de Docker para los trabajos de Dataflow

1. Regresa a la sesión de Cloud Shell para configurar las bibliotecas de Python requeridas.

Los trabajos de Dataflow de este lab requieren Python3.8. Para asegurarte de utilizar la versión correcta, ejecutarás los procesos de Dataflow en un contenedor de Docker con Python 3.8.

2. Ejecuta lo siguiente en Cloud Shell para iniciar un contenedor de Python:

cd ~ docker run -it -e PROJECT=$PROJECT -v $(pwd)/dataflow-python-examples:/dataflow python:3.8 /bin/bash

Con este comando, se extrae un contenedor de Docker con la versión estable más reciente de Python 3.8 y se activa una shell de comandos para ejecutar los siguientes comandos en el contenedor. La marca -v proporciona el código fuente como un volumen para el contenedor, de manera que podamos utilizar el editor de Cloud Shell y, aun así, acceder a él dentro del contenedor en ejecución.

3. Una vez que el contenedor termine de extraerse y comience a activarse en Cloud Shell, ejecuta lo siguiente para instalar apache-beam en él:

pip install apache-beam[gcp]==2.59.0

4. Luego, en el contenedor en ejecución en Cloud Shell, cambia al directorio en el que vinculaste el código fuente:

cd dataflow/

5. Configura el ID del proyecto en el contenedor:

export PROJECT={{{ project_0.project_id }}}

Ejecuta la canalización de transferencia de datos en la nube

1. Ejecuta el siguiente código para activar la canalización de transferencia de datos:

python dataflow_python_examples/data_ingestion.py \ --project=$PROJECT \ --region={{{ project_0.default_region | REGION }}} \ --runner=DataflowRunner \ --machine_type=e2-standard-2 \ --staging_location=gs://$PROJECT/test \ --temp_location gs://$PROJECT/test \ --input gs://$PROJECT/data_files/head_usa_names.csv \ --save_main_session

Este código inicia los trabajadores necesarios y, luego, los cierra cuando se completa la canalización.

2. En la barra de título de la consola, escribe Dataflow en el campo Buscar y, luego, haz clic en Dataflow en los resultados de la búsqueda.

Cuando se abra la página de Dataflow, consulta el estado del trabajo.

3. Haz clic en el nombre del trabajo para ver su progreso.

Una vez que el Estado del trabajo sea Sin errores, puedes continuar con el siguiente paso. Esta canalización de transferencia tarda aproximadamente cinco minutos en iniciarse, completar el trabajo y, luego, cerrarse.

4. Navega a BigQuery (menú de navegación > BigQuery) para verificar si tus datos se propagaron.

5. Haz clic en el nombre de tu proyecto para ver la tabla usa_names en el conjunto de datos lake.

6. Haz clic en la tabla y, luego, navega hasta la pestaña Vista previa para ver ejemplos de los datos de usa_names.

Nota: Si no ves la tabla usa_names, actualiza la página o visualiza las tablas en la IU clásica de BigQuery.

Haz clic en Revisar mi progreso para verificar la tarea realizada.

Crear una canalización de transferencia de datos

Tarea 6: Revisa y ejecuta la canalización de transformación de datos

En esta tarea, revisarás la canalización de transformación de datos para aprender cómo funciona. Luego, la ejecutarás para que procese los archivos de Cloud Storage y guarde el resultado en BigQuery.

Esta canalización también transfiere datos de Cloud Storage a la tabla de BigQuery con la fuente TextIO y el destino BigQueryIO, pero con transformaciones adicionales. En particular, hace lo siguiente:

• Transferir los archivos desde Cloud Storage
• Convertir las líneas leídas en objetos de diccionario
• Transformar los datos que contengan el año a un formato que BigQuery entienda como fecha
• Guardar las filas como resultados en BigQuery

Revisa el código de Python de la canalización de transformación de datos

• En el Editor de código, abre el archivo data_transformation.py.

Lee los comentarios del archivo, en los que se explica lo que hace el código.

Ejecuta la canalización de transformación de datos en la nube

1. Ejecuta el siguiente código para ejecutar la canalización de transformación de datos:

python dataflow_python_examples/data_transformation.py \ --project=$PROJECT \ --region={{{ project_0.default_region | REGION }}} \ --runner=DataflowRunner \ --machine_type=e2-standard-2 \ --staging_location=gs://$PROJECT/test \ --temp_location gs://$PROJECT/test \ --input gs://$PROJECT/data_files/head_usa_names.csv \ --save_main_session

2. En la barra de título de la consola de Google Cloud, escribe Dataflow en el campo Buscar y, luego, haz clic en Dataflow en los resultados.

3. Haz clic en el nombre del trabajo para ver su estado.

Esta canalización de Dataflow tarda aproximadamente cinco minutos en iniciarse, completar el trabajo y, luego, cerrarse.

4. Una vez que el Estado del trabajo sea Sin errores en la pantalla de estado del trabajo de Dataflow, navega a BigQuery para verificar que se hayan propagado los datos.

Deberías ver la tabla usa_names_transformed debajo del conjunto de datos lake.

5. Haz clic en la tabla y navega hasta la pestaña Vista previa para ver ejemplos de los datos de usa_names_transformed.

Nota: Si no ves la tabla usa_names_transformed, actualiza la página o visualiza las tablas por medio de la IU clásica de BigQuery.

Haz clic en Revisar mi progreso para verificar la tarea realizada.

Crear una canalización de transformación de datos

Tarea 7: Revisa y ejecuta la canalización de enriquecimiento de datos

Ahora, crearás una canalización de enriquecimiento de datos que hace lo siguiente:

• Transferir los archivos desde Cloud Storage
• Filtrar la fila del encabezado en los archivos
• Convertir las líneas leídas en objetos de diccionario
• Guardar las filas como resultados en BigQuery

Revisa y edita el código de Python de la canalización de enriquecimiento de datos

1. En el Editor de código, abre el archivo data_enrichment.py.

2. Revisa los comentarios, en los que se explica lo que hace el código (propagar datos en BigQuery).

La línea 83 actualmente se ve así:

values = [x.decode('utf8') for x in csv_row]

3. Edítala para que se vea así:

values = [x for x in csv_row]

4. Cuando termines de editar esta línea, recuerda guardar el archivo actualizado. Para ello, selecciona la opción File en el Editor de código y haz clic en Save.

Ejecuta la canalización de enriquecimiento de datos

1. Ejecuta el siguiente código para ejecutar la canalización de enriquecimiento de datos:

python dataflow_python_examples/data_enrichment.py \ --project=$PROJECT \ --region={{{ project_0.default_region | REGION }}} \ --runner=DataflowRunner \ --machine_type=e2-standard-2 \ --staging_location=gs://$PROJECT/test \ --temp_location gs://$PROJECT/test \ --input gs://$PROJECT/data_files/head_usa_names.csv \ --save_main_session

2. En la página de Dataflow, haz clic en el trabajo para ver su Estado.

Esta canalización de Dataflow tarda aproximadamente cinco minutos en iniciarse, completar el trabajo y, luego, cerrarse.

3. Una vez que el Estado del trabajo sea Sin errores en la pantalla Estado del trabajo de Dataflow, en la consola, haz clic en el menú de navegación > BigQuery para verificar que se hayan propagado los datos.

Deberías ver la tabla usa_names_enriched debajo del conjunto de datos lake.

4. Haz clic en la tabla y navega hasta la pestaña Vista previa para ver ejemplos de los datos de usa_names_enriched.

Nota: Si no ves la tabla usa_names_enriched, actualiza la página o visualiza las tablas en la IU clásica de BigQuery.

Haz clic en Revisar mi progreso para verificar la tarea realizada.

Crear una canalización de enriquecimiento de datos

Tarea 8: Revisa y ejecuta la canalización de data lake a data mart

Ahora, crearás una canalización de Dataflow que lee los datos de dos fuentes de BigQuery y, luego, las une. Específicamente, harás lo siguiente:

• Transferir archivos desde dos fuentes de BigQuery
• Unir ambas fuentes de datos
• Filtrar la fila del encabezado en los archivos
• Convertir las líneas leídas en objetos de diccionario
• Guardar las filas como resultados en BigQuery

Ejecuta la canalización de transferencia de datos para realizar una unión de datos y escribir la tabla resultante en BigQuery

Primero, revisa el código de data_lake_to_mart.py para comprender qué hace. Luego, ejecuta la canalización en la nube.

1. En el Editor de código, abre el archivo data_lake_to_mart.py.

Lee los comentarios del archivo, en los que se explica lo que hace el código (unir dos tablas y escribir los resultados en una tabla nueva en BigQuery).

2. Ejecuta el siguiente bloque de código para ejecutar la canalización:

python dataflow_python_examples/data_lake_to_mart.py \ --worker_disk_type="compute.googleapis.com/projects//zones//diskTypes/pd-ssd" \ --max_num_workers=4 \ --project=$PROJECT \ --runner=DataflowRunner \ --machine_type=e2-standard-2 \ --staging_location=gs://$PROJECT/test \ --temp_location gs://$PROJECT/test \ --save_main_session \ --region={{{ project_0.default_region | REGION }}}

3. En la barra de título de la consola de Google Cloud, escribe Dataflow en el campo Buscar y, luego, haz clic en Dataflow en los resultados.

4. Haz clic en este trabajo nuevo para ver el estado.

Esta canalización de Dataflow tarda aproximadamente cinco minutos en iniciarse, completar el trabajo y, luego, cerrarse.

5. Una vez que el Estado del trabajo sea Sin errores en la pantalla de estado del trabajo de Dataflow, haz clic en el menú de navegación > BigQuery para verificar que se hayan propagado los datos.

Deberías ver la tabla orders_denormalized_sideinput debajo del conjunto de datos lake.

6. Haz clic en la tabla y navega a la sección Vista previa para ver ejemplos de datos de orders_denormalized_sideinput.

Nota: Si no ves la tabla orders_denormalized_sideinput, actualiza la página o visualiza las tablas en la IU clásica de BigQuery.

Haz clic en Revisar mi progreso para verificar la tarea realizada.

Crear una canalización de data lake a data mart de Dataflow

Pon a prueba tus conocimientos

A continuación, encontrarás una pregunta de opción múltiple para reforzar tus conocimientos sobre los conceptos de este lab. Trata de responderla lo mejor posible.

¡Felicitaciones!

Ejecutaste código de Python con Dataflow para transferir datos de Cloud Storage a BigQuery y, luego, transformarlos y enriquecerlos en BigQuery.

Próximos pasos y más información

Si buscas más información, consulta la siguiente documentación oficial:

• Dataflow
• BigQuery
• Revisa la Guía de programación de Apache Beam si quieres consultar conceptos avanzados.
• Consulta los siguientes labs:
  • Predice las compras de visitantes con un modelo de clasificación en BigQuery ML
  • Cloud Composer: Cómo copiar tablas de BigQuery en diferentes ubicaciones

Capacitación y certificación de Google Cloud

Recibe la formación que necesitas para aprovechar al máximo las tecnologías de Google Cloud. Nuestras clases incluyen habilidades técnicas y recomendaciones para ayudarte a avanzar rápidamente y a seguir aprendiendo. Para que puedas realizar nuestros cursos cuando más te convenga, ofrecemos distintos tipos de capacitación de nivel básico a avanzado: a pedido, presenciales y virtuales. Las certificaciones te ayudan a validar y demostrar tus habilidades y tu conocimiento técnico respecto a las tecnologías de Google Cloud.

Última actualización del manual: 1 de abril de 2025

Prueba más reciente del lab: 1 de abril de 2025

Copyright 2025 Google LLC. All rights reserved. Google y el logotipo de Google son marcas de Google LLC. Los demás nombres de productos y empresas pueden ser marcas de las respectivas empresas a las que estén asociados.