GSP283

Descripción general

Imagina que tienes conjuntos de datos en diferentes ubicaciones del mundo y que estos datos se encuentran en buckets de Google Cloud Storage o en tablas de BigQuery. ¿Cómo puedes organizarlos para que se consoliden y analicen para proporcionar estadísticas sobre tu negocio?

Cloud Composer puede ayudarte a crear flujos de trabajo y trasladar tus datos entre regiones y sistemas de almacenamiento con una vista gráfica intuitiva. Entre otros beneficios, cuenta con plantillas para transferir datos entre BigQuery y Cloud Storage de forma sencilla y confiable.

En este lab, crearás y ejecutarás un flujo de trabajo de Apache Airflow en Cloud Composer que completará las siguientes tareas:

• Leer un archivo de configuración que proporciona una lista de tablas para copiar
• Exportar a Cloud Storage la lista de tablas de un conjunto de datos de BigQuery ubicado en EE.UU.
• Copiar las tablas exportadas del bucket de Cloud Storage de EE.UU. a la UE
• Importar la lista de tablas al conjunto de datos de destino de BigQuery de la UE

Actividades

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

• Crear un entorno de Cloud Composer
• Crear buckets de Cloud Storage
• Crear conjuntos de datos de BigQuery
• Crear y ejecutar flujos de trabajo de Apache Airflow en Cloud Composer para mover datos entre Cloud Storage y BigQuery

Configuración y requisitos

Antes de hacer clic en el botón Comenzar lab

Lee estas instrucciones. Los labs son cronometrados y no se pueden pausar. El cronómetro, 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 de navegador privada o de Incógnito para ejecutar este lab. Así evitarás cualquier conflicto 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: Si ya tienes un proyecto o una cuenta personal de Google Cloud, no los uses en este lab para evitar cargos adicionales en tu cuenta.

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

1. Haga clic en el botón Comenzar lab. Si debe pagar por el lab, se abrirá una ventana emergente para que seleccione su forma de pago. A la izquierda, se encuentra el panel Detalles del lab, que tiene estos elementos:

   • El botón Abrir la consola de Google Cloud
   • El tiempo restante
   • Las credenciales temporales que debe 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: Ordene 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 podría generar cargos adicionales.

7. Haga 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 ver un menú con una lista de productos y servicios de Google Cloud, haz clic en el menú de navegación que se encuentra en la parte superior izquierda.

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.

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 YOUR_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.

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

gcloud auth list

3. Haz clic en Autorizar.

4. Ahora, el resultado debería verse de la siguiente manera:

Resultado:

ACTIVE: * ACCOUNT: student-01-xxxxxxxxxxxx@qwiklabs.net 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_ID>

Resultado de ejemplo:

[core] project = qwiklabs-gcp-44776a13dea667a6 Nota: Para obtener toda la documentación de gcloud, consulta la guía con la descripción general de gcloud CLI en Google Cloud.

Tarea 1. Crea un entorno de Cloud Composer

1. Para crear un entorno de Cloud Composer, escribe Composer en el campo de búsqueda de la barra del título de la consola de Google Cloud y, luego, haz clic en Composer en la sección Productos y páginas.

2. Luego, haz clic en Crear entorno.

3. En el menú desplegable, selecciona Composer 3.

4. Configura los siguientes parámetros para tu entorno:

• Nombre: composer-advanced-lab

• Ubicación:

• Versión de la imagen: composer-3-airflow-n.n.n-build.n (selecciona la imagen con el número más alto que esté disponible)

• En Recursos del entorno, selecciona Pequeño.

• Haz clic en el menú desplegable de Mostrar configuración avanzada y selecciona Zona de la base de datos de Airflow como .

Deja el resto de la configuración con sus valores predeterminados.

5. Haz clic en Crear.

El proceso de creación del entorno se completa cuando, en la página Entornos de la consola de Cloud, la marca de verificación verde aparece a la izquierda del nombre del entorno.

Nota: El entorno puede demorar hasta 20 minutos en completar el proceso de configuración. Continúa con la siguiente sección: Crea buckets de Cloud Storage y conjuntos de datos de destino de BigQuery.

Haz clic en Revisar mi progreso para verificar el objetivo.

Crear el entorno de Cloud Composer

Tarea 2. Crea buckets de Cloud Storage

En esta tarea, crearás dos buckets multirregionales de Cloud Storage. Estos buckets se usarán para copiar las tablas exportadas entre las distintas ubicaciones; p. ej., de EE.UU. a la UE.

Crea un bucket en EE.UU.

1. Ve a Cloud Storage > Bucket y haz clic en Crear.
2. Asígnale al bucket un nombre único que incluya el ID del proyecto (p. ej., -us).
3. Para el Tipo de ubicación, selecciona us (varias regiones en Estados Unidos).
4. Deja las demás opciones con sus valores predeterminados y haz clic en Crear.
5. Marca la casilla Aplicar la prevención de acceso público a este bucket y haz clic en Confirmar en la ventana emergente Se impedirá el acceso público si se solicita.

Crea un bucket en la UE

Repite los pasos para crear otro bucket en la región EU. El nombre único debe incluir la ubicación como sufijo de tu bucket (p. ej., -eu).

Haz clic en Revisar mi progreso para verificar el objetivo.

Crear dos buckets de Cloud Storage

Tarea 3. Crea el conjunto de datos de destino de BigQuery

1. Crea el conjunto de datos de destino de BigQuery en la UE desde la nueva IU web de BigQuery.

2. Ve a Menú de navegación > BigQuery.

Se abrirá el cuadro de mensaje Te damos la bienvenida a BigQuery en la consola de Cloud, que contiene un vínculo a la guía de inicio rápido y enumera las actualizaciones de la IU.

3. Haz clic en Listo.

4. Luego, haz clic en los tres puntos junto al ID del proyecto de Qwiklabs y selecciona Crear conjunto de datos.

5. Usa el ID del conjunto de datos nyc_tlc_EU, para el tipo de ubicación, elige Multirregional y selecciona UE en el menú desplegable.

6. Haz clic en CREAR CONJUNTO DE DATOS.

Haz clic en Revisar mi progreso para verificar el objetivo.

Crear un conjunto de datos

Tarea 4: Airflow y conceptos básicos: una breve introducción

• Mientras desarrollas tu entorno, consulta el archivo de muestra que utilizarás en este lab.

Airflow es una plataforma para crear, programar y supervisar flujos de trabajo de forma programática.

Utiliza Airflow para crear flujos de trabajo como grafos acíclicos dirigidos (DAG) de tareas. El programador de Airflow ejecuta tus tareas en un array de trabajadores mientras sigue las dependencias especificadas.

Conceptos básicos

DAG: Un grafo acíclico dirigido es una colección de tareas organizadas para reflejar sus relaciones y dependencias.

Operador: Es la descripción de una sola tarea; suele ser atómico. Por ejemplo, BashOperator se usa para ejecutar el comando Bash.

Tarea: Es una instancia con parámetros de un operador; un nodo en el DAG.

Instancia de tarea: Es la ejecución específica de una tarea, caracterizada como un DAG, una tarea y un momento determinado. Tiene un estado indicativo: running, success, failed, skipped, …

Obtén más información sobre los conceptos de Airflow en la documentación de los conceptos.

Tarea 5. Define el flujo de trabajo

Los flujos de trabajo de Cloud Composer constan de DAG (grafos acíclicos dirigidos). El código que aparece en bq_copy_across_locations.py es el código del flujo de trabajo, también conocido como DAG. Abre el archivo para ver cómo se conforma. Luego, podrás conocer algunos de los componentes clave del archivo.

Para organizar todas las tareas del flujo de trabajo, el DAG importa los siguientes operadores:

1. DummyOperator: Crea tareas iniciales y finales ficticias para una mejor representación visual del DAG.
2. BigQueryToCloudStorageOperator: Exporta tablas de BigQuery a buckets de Cloud Storage con el formato Avro.
3. GoogleCloudStorageToGoogleCloudStorageOperator: Copia los archivos entre los buckets de Cloud Storage.
4. GoogleCloudStorageToBigQueryOperator: Importa tablas desde archivos Avro al bucket de Cloud Storage.

• En este ejemplo, se define la función read_master_file() de modo que se pueda leer el archivo de configuración y crear la lista de tablas para copiarla:

# -------------------------------------------------------------------------------- # Functions # -------------------------------------------------------------------------------- def read_table_list(table_list_file): """ Reads the master CSV file that will help in creating Airflow tasks in the DAG dynamically. :param table_list_file: (String) The file location of the master file, e.g. '/home/airflow/framework/master.csv' :return master_record_all: (List) List of Python dictionaries containing the information for a single row in master CSV file. """ master_record_all = [] logger.info('Reading table_list_file from : %s' % str(table_list_file)) try: with open(table_list_file, 'rb') as csv_file: csv_reader = csv.reader(csv_file) next(csv_reader) # skip the headers for row in csv_reader: logger.info(row) master_record = { 'table_source': row[0], 'table_dest': row[1] } master_record_all.append(master_record) return master_record_all except IOError as e: logger.error('Error opening table_list_file %s: ' % str( table_list_file), e)

• El nombre del DAG es bq_copy_us_to_eu_01. No se programó el DAG de forma predeterminada, de modo que se lo debe activar manualmente.

default_args = { 'owner': 'airflow', 'start_date': datetime.today(), 'depends_on_past': False, 'email': [''], 'email_on_failure': False, 'email_on_retry': False, 'retries': 1, 'retry_delay': timedelta(minutes=5), } # DAG object. with models.DAG('bq_copy_us_to_eu_01', default_args=default_args, schedule_interval=None) as dag:

• Para definir el complemento de Cloud Storage, se define la clase Cloud StoragePlugin(AirflowPlugin) y se asignan el hook y el operador descargados desde la rama 1.10-stable de Airflow.

# Import operator from plugins from gcs_plugin.operators import gcs_to_gcs

Tarea 6. Ve la información del entorno

1. Vuelve a Composer para comprobar el estado de tu entorno.

2. Una vez que se haya creado tu entorno, haz clic en el nombre para ver sus detalles.

La página Detalles del entorno brinda información como la URL de la IU web de Airflow, el ID de clúster de Google Kubernetes Engine y el nombre del bucket de Cloud Storage conectado a la carpeta de DAG.

Nota: Cloud Composer usa Cloud Storage para almacenar los DAG de Apache Airflow, también conocidos como flujos de trabajo. Cada entorno está asociado a un bucket de Cloud Storage. Cloud Composer programa solo los DAG del bucket de Cloud Storage.

Se deben completar los siguientes pasos en Cloud Shell.

Crea un entorno virtual

Se utilizan entornos virtuales de Python para aislar del sistema la instalación de paquetes.

1. Instala el entorno virtualenv:

sudo apt-get install -y virtualenv

2. Crea el entorno virtual:

python3 -m venv venv

3. Activa el entorno virtual.

source venv/bin/activate

Tarea 7. Crea una variable para el bucket de DAG de Cloud Storage

• En Cloud Shell, ejecuta lo siguiente para copiar el nombre del bucket de DAG que se encuentra en tu página Detalles del entorno y establecer una variable para hacer referencia a ella en Cloud Shell:

Nota: Asegúrate de reemplazar el nombre de tu bucket de DAG en el siguiente comando. Abre el Menú de navegación > Cloud Storage. Se verá similar a -composer-advanced-YOURDAGSBUCKET-bucket (YOURDAGSBUCKET-bucket). DAGS_BUCKET=<your DAGs bucket name>

En este lab, deberás usar esta variable algunas veces.

Tarea 8. Configura variables de Airflow

Las variables de Airflow son un concepto específico de Airflow que difiere de las variables de entorno. En este paso, establecerás las siguientes tres variables de Airflow que utiliza el DAG que implementaremos: table_list_file_path, gcs_source_bucket y gcs_dest_bucket.

Clave	Valor	Detalles
table_list_file_path	/home/airflow/gcs/dags/bq_copy_eu_to_us_sample.csv	Archivo CSV que indica las tablas de origen y destino, incluido el conjunto de datos
gcs_source_bucket	{ID ÚNICO}-us	Bucket de Cloud Storage para exportar tabledest_bbucks de BigQuery desde el origen
gcs_dest_bucket	{ID ÚNICO}-eu	Bucket de Cloud Storage para importar tablas de BigQuery en el destino

El siguiente comando de gcloud composer ejecuta las variables del subcomando de la CLI de Airflow. El subcomando pasa los argumentos a la herramienta de línea de comandos de gcloud.

Para establecer las tres variables, ejecutarás composer command una vez por cada fila de la tabla anterior. Esta es la forma del comando:

gcloud composer environments run ENVIRONMENT_NAME \ --location LOCATION variables -- \ set KEY VALUE Puedes ignorar de forma segura este error de gcloud: (ERROR: gcloud crashed (TypeError): 'NoneType' object is not callable). Este es un problema conocido relacionado con el uso de entornos de Composer de gcloud que se ejecutaron con la versión 410.0.0 de gcloud. Tus variables se definirán según corresponda así te aparezca este mensaje de error.

• ENVIRONMENT_NAME es el nombre del entorno.
• LOCATION es la región de Compute Engine en la que se encuentra el entorno. Para el comando de gcloud composer, debes incluir la marca --location o configurar la ubicación predeterminada antes de ejecutar el comando de gcloud.
• KEY y VALUE especifican la variable y su valor para configurarla. Incluye un espacio, dos guiones y un espacio ( -- ) entre los argumentos relacionados con el comando de gcloud a la izquierda y los argumentos relacionados con el subcomando de Airflow a la derecha. Además, incluye un espacio entre los argumentos de KEY y VALUE; usa el comando gcloud composer environments run con el subcomando de las variables.

Ejecuta estos comandos en Cloud Shell. Para ello, reemplaza gcs_source_bucket y gcs_dest_bucket por los nombres de los buckets que creaste en la tarea 2.

gcloud composer environments run composer-advanced-lab \ --location {{{ project_0.default_region | "REGION" }}} variables -- \ set table_list_file_path /home/airflow/gcs/dags/bq_copy_eu_to_us_sample.csv gcloud composer environments run composer-advanced-lab \ --location {{{ project_0.default_region | "REGION" }}} variables -- \ set gcs_source_bucket {UNIQUE ID}-us gcloud composer environments run composer-advanced-lab \ --location {{{ project_0.default_region | "REGION" }}} variables -- \ set gcs_dest_bucket {UNIQUE_ID}-eu

Para ver el valor de una variable, ejecuta las variables de subcomando de la CLI de Airflow con el argumento get o usa la IU de Airflow.

Por ejemplo, ejecuta lo siguiente:

gcloud composer environments run composer-advanced-lab \ --location {{{ project_0.default_region | "REGION" }}} variables -- \ get gcs_source_bucket Nota: Asegúrate de definir las tres variables de Airflow que usa el DAG.

Tarea 9. Sube el DAG y las dependencias a Cloud Storage

1. Copia los archivos de muestra de los documentos de Python de Google Cloud en Cloud Shell:

cd ~ gcloud storage cp -r gs://spls/gsp283/python-docs-samples .

2. Sube una copia del hook y del operador de terceros a la carpeta de complementos de tu bucket de DAG de Composer perteneciente a Cloud Storage:

gcloud storage cp -r python-docs-samples/third_party/apache-airflow/plugins/* gs://$DAGS_BUCKET/plugins

3. Luego, sube el DAG y el archivo de configuración al bucket de DAG de Cloud Storage de tu entorno:

gcloud storage cp python-docs-samples/composer/workflows/bq_copy_across_locations.py gs://$DAGS_BUCKET/dags gcloud storage cp python-docs-samples/composer/workflows/bq_copy_eu_to_us_sample.csv gs://$DAGS_BUCKET/dags

Cloud Composer registra automáticamente el DAG en tu entorno de Airflow, y los cambios en el DAG tardan de 3 a 5 minutos. Puedes ver el estado de la tarea en la interfaz web de Airflow y confirmar que el DAG no esté programado según la configuración.

Tarea 10. Explora la IU de Airflow

Sigue estos pasos para acceder a la interfaz web de Airflow con la consola de Cloud:

1. Regresa a la página Entornos de Composer.
2. En la columna Webserver de Airflow del entorno, haz clic en el vínculo Airflow.

3. Haz clic en tus credenciales del lab.
4. Se abrirá la IU web de Airflow en una nueva ventana del navegador. Seguirán cargándose los datos cuando llegues a este punto. Mientras tanto, puedes continuar con el lab.

Visualiza las variables

Las variables que configuraste anteriormente se conservan en tu entorno.

• Para visualizar las variables, selecciona Administrador > Variables en la barra de menú de Airflow.

Cómo activar el DAG para que se ejecute de forma manual

1. Haz clic en la pestaña DAG y espera a que se terminen de cargar los vínculos.

2. Para activar el DAG de forma manual, haz clic en el botón de reproducción de composer_sample_bq_copy_across_locations:

3. Haz clic en Activar DAG para confirmar esta acción.

Haz clic en Revisar mi progreso para verificar el objetivo.

Subir el DAG y las dependencias a Cloud Storage

Cómo explorar las ejecuciones del DAG

Cuando subas tu archivo DAG a la carpeta de DAG en Cloud Storage, Cloud Composer analizará el archivo. Si no se detectan errores, aparecerá el nombre del flujo de trabajo en la lista de DAG, y se pondrá el flujo de trabajo en cola para que se lo ejecute inmediatamente, siempre y cuando se cumplan las condiciones del programa (en este caso, la configuración debe ser “Ninguna”).

El estado de las ejecuciones se pondrá verde cuando presiones el botón de reproducción:

1. Haz clic en el nombre del DAG para abrir su página de detalles. En esta página, se incluye una representación gráfica de las dependencias y tareas del flujo de trabajo.

2. En la barra de herramientas, haz clic en Grafo y desplaza el mouse sobre el gráfico de cada tarea para conocer su estado. Ten en cuenta que el borde de cada tarea también indica el estado (borde verde: running; rojo: failed; etcétera).

Para volver a ejecutar el flujo de trabajo desde la vista del Grafo, haz lo siguiente:

1. En la Vista de gráfico de la IU de Airflow, haz clic en el gráfico de inicio.
2. Haz clic en Borrar para restablecer todas las tareas y, luego, en Aceptar para confirmar.

Actualiza el navegador mientras se ejecuta el proceso para ver la información más reciente.

Tarea 11. Valida los resultados

Para comprobar el estado y los resultados del flujo de trabajo, dirígete a estas páginas de la consola de Cloud:

• Se copiaron las tablas exportadas del bucket de EE.UU. al de Cloud Storage de la UE. Haz clic en Cloud Storage para ver los archivos Avro intermedios en los buckets de origen (EE.UU.) y destino (UE).
• Se importó la lista de tablas al conjunto de datos de destino de BigQuery. Haz clic en BigQuery y, luego, en el nombre de tu proyecto y en el conjunto de datos nyc_tlc_EU para asegurarte de poder acceder a las tablas desde el conjunto de datos que creaste.

Borre el entorno de Cloud Composer

1. Vuelva a la página de Entornos en Compositor.
2. Seleccione la casilla junto a su entorno de Compositor.
3. Haga clic en BORRAR.
4. Confirme la ventana emergente haciendo clic en BORRAR otra vez.

¡Felicitaciones!

Copiaste las tablas de forma programática de EE.UU. a la UE. Este lab se basa en esta entrada de blog de David Sabater Dinter.

Próximos pasos

• Obtén más información para usar Airflow en el sitio web de Airflow o en el proyecto de GitHub de Airflow.
• Existen muchos otros recursos disponibles para Airflow, como el grupo de discusión.
• Regístrate para crear una cuenta de JIRA para Apache y reabre cualquier problema que sea de tu interés en el proyecto de JIRA de Apache Airflow.
• Para obtener información sobre la IU de Airflow, consulta Accede a la interfaz web.

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: 21 de junio de 2024

Prueba más reciente del lab: 21 de junio de 2024

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.