Descripción general

Los flujos de trabajo son un tema recurrente en el análisis de datos. Implican transferir, transformar y analizar datos para descubrir la información significativa que contienen. En Google Cloud, la herramienta destinada a alojar flujos de trabajo es Cloud Composer, que es una versión alojada de Apache Airflow, la popular herramienta de código abierto para flujos de trabajo.

En este lab, usarás la consola de Google Cloud para configurar un entorno de Cloud Composer. Luego, utilizarás Cloud Composer para preparar un flujo de trabajo simple que verifique la existencia de un archivo de datos, cree un clúster de Cloud Dataproc, ejecute un trabajo de conteo de palabras de Apache Hadoop en el clúster de Cloud Dataproc y borre el clúster más adelante.

Actividades

• Usar la consola de Google Cloud para crear el entorno de Cloud Composer

• Ver y ejecutar el DAG (grafo acíclico dirigido) en la interfaz web de Airflow

• Ver los resultados del trabajo de recuento de palabras en Storage

Configuración y requisitos

Configuración del lab

En cada lab, recibirá un proyecto de Google Cloud y un conjunto de recursos nuevos por tiempo limitado y sin costo adicional.

1. Accede a Qwiklabs desde una ventana de incógnito.

2. Ten en cuenta el tiempo de acceso del lab (por ejemplo, 1:15:00) y asegúrate de finalizarlo en el plazo asignado.
   No existe una función de pausa. Si lo necesita, puede reiniciar el lab, pero deberá hacerlo desde el comienzo.

3. Cuando esté listo, haga clic en Comenzar lab.

4. Anote las credenciales del lab (el nombre de usuario y la contraseña). Las usarás para acceder a la consola de Google Cloud.

5. Haga clic en Abrir Google Console.

6. Haga clic en Usar otra cuenta, copie las credenciales para este lab y péguelas en el mensaje emergente que aparece.
   Si usa otras credenciales, se generarán errores o incurrirá en cargos.

7. Acepta las condiciones y omite la página de recursos de recuperación.

Activa Google Cloud Shell

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

Google Cloud Shell proporciona acceso de línea de comandos a tus recursos de Google Cloud.

1. En la consola de Cloud, en la barra de herramientas superior derecha, haz clic en el botón Abrir Cloud Shell.

   

2. Haz clic en Continuar.

El aprovisionamiento y la conexión al entorno demorarán unos minutos. Cuando te conectes, habrás completado la autenticación, y el proyecto estará configurado con tu PROJECT_ID. Por ejemplo:

gcloud es la herramienta de línea de comandos de Google Cloud. Viene preinstalada en Cloud Shell y es compatible con el completado de línea de comando.

• Puedes solicitar el nombre de la cuenta activa con este comando:

gcloud auth list

Resultado:

Credentialed accounts: - @.com (active)

Resultado de ejemplo:

Credentialed accounts: - google1623327_student@qwiklabs.net

• Puedes solicitar el ID del proyecto con este comando:

gcloud config list project

Resultado:

[core] project =

Resultado de ejemplo:

[core] project = qwiklabs-gcp-44776a13dea667a6 Nota: La documentación completa de gcloud está disponible en la guía de descripción general de gcloud CLI .

Verifica los permisos del proyecto

Antes de comenzar a trabajar en Google Cloud, asegúrate de que tu proyecto tenga los permisos correctos en Identity and Access Management (IAM).

1. En la consola de Google Cloud, en el Menú de navegación (), selecciona IAM y administración > IAM.

2. Confirma que aparezca la cuenta de servicio predeterminada de Compute {número-del-proyecto}-compute@developer.gserviceaccount.com, y que tenga asignado el rol Editor. El prefijo de la cuenta es el número del proyecto, que puedes encontrar en el Menú de navegación > Descripción general de Cloud > Panel.

Nota: Si la cuenta no aparece en IAM o no tiene asignado el rol Editor, sigue los pasos que se indican a continuación para asignar el rol necesario.

1. En la consola de Google Cloud, en el Menú de navegación, haz clic en Descripción general de Cloud > Panel.
2. Copia el número del proyecto (p. ej., 729328892908).
3. En el Menú de navegación, selecciona IAM y administración > IAM.
4. En la parte superior de la tabla de funciones, debajo de Ver por principales, haz clic en Otorgar acceso.
5. En Principales nuevas, escribe lo siguiente:

{project-number}-compute@developer.gserviceaccount.com

6. Reemplaza {número-del-proyecto} por el número de tu proyecto.
7. En Rol, selecciona Proyecto (o Básico) > Editor.
8. Haz clic en Guardar.

Tarea 1. Tareas del entorno

1. En la barra de título de la consola de Google Cloud, haz clic en Activar Cloud Shell. Si se te solicita, haz clic en Continuar.

2. Ejecuta los siguientes comandos para asignar el rol de Trabajador de Composer a la cuenta de servicio de desarrollador de Compute.

export PROJECT_ID=$(gcloud config get-value project) export PROJECT_NUMBER=$(gcloud projects describe $PROJECT_ID --format="value(projectNumber)") gcloud projects add-iam-policy-binding {{{project_0.project_id}}} \ --member=serviceAccount:$PROJECT_NUMBER-compute@developer.gserviceaccount.com \ --role=roles/composer.worker

3. Ejecuta los siguientes comandos para asegurarte de que las APIs necesarias estén habilitadas correctamente en tu proyecto.

gcloud services disable composer.googleapis.com gcloud services disable artifactregistry.googleapis.com gcloud services disable container.googleapis.com gcloud services enable artifactregistry.googleapis.com gcloud services enable container.googleapis.com gcloud services enable composer.googleapis.com

Tarea 2: Crea el entorno de Cloud Composer

En esta sección, crearás un entorno de Cloud Composer.

Nota: Antes de continuar, comprueba que realizaste las tareas anteriores para garantizar que las APIs necesarias estén habilitadas correctamente. De lo contrario, la creación del entorno de Cloud Composer fallará.

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

2. Haz clic en Crear entorno y selecciona Composer 3. Establece lo siguiente para el entorno:

Propiedad	Valor
Nombre	highcpu
Ubicación
Versión de la imagen	composer-3-airflow-n.n.n-build.n (Nota: Selecciona la imagen con el número más alto que esté disponible)

3. En Recursos del entorno, selecciona Pequeño.

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

4. Haz clic en Crear.

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

El entorno puede tardar entre 15 y 30 minutos en completar el proceso de configuración. Continúa con el lab mientras se inicia.

Haz clic en Revisar mi progreso para verificar el objetivo.

Crear el entorno de Cloud Composer.

Crea un bucket de Cloud Storage

Crea un bucket de Cloud Storage en tu proyecto. Este bucket almacenará el resultado del trabajo de Hadoop obtenido de Dataproc.

1. Ve al Menú de navegación > Cloud Storage > Buckets y, luego, haz clic en + Crear.

2. Asígnale al bucket un nombre único a nivel universal, como el ID de tu proyecto () y, luego, haz clic en Crear. Si aparece el mensaje Se impedirá el acceso público, haz clic en Confirmar.

Recuerda el nombre del bucket de Cloud Storage para usarlo como variable de Airflow más adelante en el lab.

Haz clic en Revisar mi progreso para verificar el objetivo.

Crear un bucket de Cloud Storage

Tarea 3: Airflow y conceptos básicos

Mientras esperas a que se cree tu entorno de Composer, repasa algunos términos que se usan con Airflow.

Airflow es una plataforma para crear, programar y supervisar flujos de trabajo de manera 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 todas las tareas que deseas ejecutar, organizadas de manera que reflejen sus relaciones y dependencias.

Operador

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

Tarea

Es una instancia parametrizada de un operador; un nodo del DAG.

Instancia de tarea

Es una ejecución específica de una tarea, caracterizada como un DAG, una tarea y un punto en el tiempo. Tiene un estado indicativo: running, success, failed, skipped…

Puedes obtener más información sobre los conceptos en la documentación correspondiente.

Tarea 4: Define el flujo de trabajo

Ahora, analicemos el flujo de trabajo que utilizarás. Los flujos de trabajo de Cloud Composer constan de DAG (grafos acíclicos dirigidos). Los DAG se definen en archivos estándar de Python ubicados en la carpeta DAG_FOLDER de Airflow. Este, a su vez, ejecutará el código en cada archivo para compilar dinámicamente los objetos DAG. Puedes tener tantos DAG como desees, pues cada uno describe un número arbitrario de tareas. En general, cada DAG debe corresponder a un único flujo de trabajo lógico.

A continuación, se encuentra el código de flujo de trabajo hadoop_tutorial.py, también denominado DAG:

"""Example Airflow DAG that creates a Cloud Dataproc cluster, runs the Hadoop wordcount example, and deletes the cluster. This DAG relies on three Airflow variables https://airflow.apache.org/concepts.html#variables * gcp_project - Google Cloud Project to use for the Cloud Dataproc cluster. * gce_zone - Google Compute Engine zone where Cloud Dataproc cluster should be created. * gce_region - Google Compute Engine region where Cloud Dataproc cluster should be created. * gcs_bucket - Google Cloud Storage bucket to used as output for the Hadoop jobs from Dataproc. See https://cloud.google.com/storage/docs/creating-buckets for creating a bucket. """ import datetime import os from airflow import models from airflow.contrib.operators import dataproc_operator from airflow.utils import trigger_rule # Output file for Cloud Dataproc job. output_file = os.path.join( models.Variable.get('gcs_bucket'), 'wordcount', datetime.datetime.now().strftime('%Y%m%d-%H%M%S')) + os.sep # Path to Hadoop wordcount example available on every Dataproc cluster. WORDCOUNT_JAR = ( 'file:///usr/lib/hadoop-mapreduce/hadoop-mapreduce-examples.jar' ) # Arguments to pass to Cloud Dataproc job. wordcount_args = ['wordcount', 'gs://pub/shakespeare/rose.txt', output_file] yesterday = datetime.datetime.combine( datetime.datetime.today() - datetime.timedelta(1), datetime.datetime.min.time()) default_dag_args = { # Setting start date as yesterday starts the DAG immediately when it is # detected in the Cloud Storage bucket. 'start_date': yesterday, # To email on failure or retry set 'email' arg to your email and enable # emailing here. 'email_on_failure': False, 'email_on_retry': False, # If a task fails, retry it once after waiting at least 5 minutes 'retries': 1, 'retry_delay': datetime.timedelta(minutes=5), 'project_id': models.Variable.get('gcp_project') } with models.DAG( 'composer_hadoop_tutorial', # Continue to run DAG once per day schedule_interval=datetime.timedelta(days=1), default_args=default_dag_args) as dag: # Create a Cloud Dataproc cluster. create_dataproc_cluster = dataproc_operator.DataprocClusterCreateOperator( task_id='create_dataproc_cluster', # Give the cluster a unique name by appending the date scheduled. # See https://airflow.apache.org/code.html#default-variables cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}', num_workers=2, region=models.Variable.get('gce_region'), zone=models.Variable.get('gce_zone'), image_version='2.0', master_machine_type='e2-standard-2', worker_machine_type='e2-standard-2') # Run the Hadoop wordcount example installed on the Cloud Dataproc cluster # master node. run_dataproc_hadoop = dataproc_operator.DataProcHadoopOperator( task_id='run_dataproc_hadoop', region=models.Variable.get('gce_region'), main_jar=WORDCOUNT_JAR, cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}', arguments=wordcount_args) # Delete Cloud Dataproc cluster. delete_dataproc_cluster = dataproc_operator.DataprocClusterDeleteOperator( task_id='delete_dataproc_cluster', region=models.Variable.get('gce_region'), cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}', # Setting trigger_rule to ALL_DONE causes the cluster to be deleted # even if the Dataproc job fails. trigger_rule=trigger_rule.TriggerRule.ALL_DONE) # Define DAG dependencies. create_dataproc_cluster >> run_dataproc_hadoop >> delete_dataproc_cluster

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

1. DataprocClusterCreateOperator: Crea un clúster de Cloud Dataproc.
2. DataProcHadoopOperator: Envía un trabajo de conteo de palabras de Hadoop y escribe los resultados en un bucket de Cloud Storage.
3. DataprocClusterDeleteOperator: Borra el clúster para no generar cargos continuos de Compute Engine.

Las tareas se ejecutan de forma secuencial, lo que puedes ver en esta sección del archivo:

# Define DAG dependencies. create_dataproc_cluster >> run_dataproc_hadoop >> delete_dataproc_cluster

El nombre del DAG es composer_hadoop_tutorial, y este se ejecuta una vez por día:

with models.DAG( 'composer_hadoop_tutorial', # Continue to run DAG once per day schedule_interval=datetime.timedelta(days=1), default_args=default_dag_args) as dag:

Dado que la start_date que se pasa a default_dag_args está establecida como yesterday, Cloud Composer programa el flujo de trabajo para que comience inmediatamente después de subir el DAG.

Tarea 5: Consulta información del entorno

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

2. Cuando termine de crearse el entorno, haz clic en el nombre (highcpu) para ver los detalles.

En la pestaña Configuración del entorno, verás información como la URL de la IU web de Airflow, el clúster de GKE y un vínculo a la carpeta de los DAG, que está almacenada en tu bucket.

Nota: Cloud Composer solo programa los flujos de trabajo en la carpeta /dags.

Tarea 6. Usa la IU de Airflow

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

1. Regresa a la página Entornos.
2. En la columna Webserver de Airflow del entorno, haz clic en Airflow.
3. Haz clic en tus credenciales del lab.
4. La interfaz web de Airflow se abrirá en una nueva ventana del navegador.

Tarea 7: Configura variables de Airflow

Las variables de Airflow son un concepto específico de Airflow que difiere de las variables de entorno.

1. En la interfaz de Airflow, selecciona Administrador > Variables en la barra de menú.

2. Haz clic en el ícono + para agregar un registro nuevo.

3. Crea las siguientes variables de Airflow: gcp_project, gcs_bucket, gce_zone y gce_region. Haz clic en Guardar después de cada variable.

Clave	Val	Detalles
gcp_project		El proyecto de Google Cloud Platform que estás usando en este lab.
gcs_bucket	gs://<my-bucket>	Reemplaza <my-bucket> por el nombre del bucket de Cloud Storage que creaste anteriormente. Este bucket almacena el resultado de los trabajos de Hadoop que se obtienen de Dataproc.
gce_zone		Esta es la zona de Compute Engine en la que se creará tu clúster de Cloud Dataproc.
gce_region		Esta es la región de Compute Engine en la que se creará tu clúster de Cloud Dataproc.

Haz clic en Guardar. Después de agregar la primera variable, repite el mismo proceso para la segunda y la tercera. Cuando termines, la tabla Variables debería verse de esta forma:

Tarea 8: Sube el DAG a Cloud Storage

Sigue estos pasos para subir el DAG:

1. En Cloud Shell, ejecuta el comando que se menciona a continuación para subir una copia del archivo hadoop_tutorial.py al bucket de Cloud Storage que se generó automáticamente cuando creaste el entorno.

2. Reemplaza <DAGs_folder_path> en el siguiente comando por la ruta a la carpeta de los DAG:

gcloud storage cp gs://cloud-training/datawarehousing/lab_assets/hadoop_tutorial.py <DAGs_folder_path>

• Puedes obtener la ruta en Composer.
• Haz clic en el entorno que creaste anteriormente y, luego, en la pestaña Environment Configuration para ver los detalles del entorno.
• Encuentra la carpeta de los DAG y copia la ruta.

El comando revisado para subir el archivo será similar al siguiente:

gcloud storage cp gs://cloud-training/datawarehousing/lab_assets/hadoop_tutorial.py gs://{{{project_0.default_region|REGION}}}-highcpu-0682d8c0-bucket/dags

3. Cuando el archivo se haya subido correctamente al directorio de DAG, abre la carpeta dags en el bucket y verás el archivo en la pestaña Objetos de los detalles del bucket.

Cuando se agrega un archivo DAG a la carpeta de los DAG, Cloud Composer agrega el DAG a Airflow y lo programa automáticamente. Los cambios en el DAG tardan de 3 a 5 minutos.

Puedes ver el estado de la tarea del DAG composer_hadoop_tutorial en la interfaz web de Airflow.

Nota: Puedes ignorar de forma segura cualquier mensaje de la interfaz, como “Parece que el programador no está ejecutándose…”. La interfaz web de Airflow se actualizará a medida que el DAG avance.

Haz clic en Revisar mi progreso para verificar el objetivo.

Subir el DAG a Cloud Storage

Explora las ejecuciones del DAG

Cuando subas tu archivo DAG a la carpeta dags en Cloud Storage, Cloud Composer lo analizará. Si no se encuentran errores, el nombre del flujo de trabajo aparecerá en la lista del DAG y el flujo de trabajo se pondrá en cola para ejecutarse de inmediato.

1. Asegúrate de estar en la pestaña de DAG, en la interfaz web de Airflow. Este proceso tardará varios minutos en completarse. Actualiza la página para asegurarte de ver la información más reciente.

2. En Airflow, haz clic en composer_hadoop_tutorial para abrir la página de detalles del DAG. En ella, se incluyen varias representaciones de las tareas y dependencias del flujo de trabajo.

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

4. Haz clic en el vínculo “Actualizar” para asegurarte de ver la información más reciente. Los bordes de los procesos cambian de color a medida que se actualizan los estados.

Nota: Si tu clúster de Dataproc ya existe, puedes volver a ejecutar el flujo de trabajo para alcanzar el estado “success” haciendo clic en el gráfico “create_dataproc_cluster” y, luego, en Borrar para restablecer las tres tareas y, por último, en Aceptar para confirmar.

5. Una vez que el estado de create_dataproc_cluster cambie a “running”, dirígete al menú de navegación > Dataproc y, luego, haz clic en lo siguiente:

   • Clústeres para supervisar la creación y eliminación de clústeres. El clúster que el flujo de trabajo creó es efímero: solo existe a la par del flujo de trabajo y se borra como parte de la última tarea de este.
   • Trabajos para supervisar el trabajo de conteo de palabras de Apache Hadoop. Haz clic en el ID del trabajo para ver el resultado del registro de trabajos.

6. Cuando Dataproc alcance el estado “Running”, vuelve a Airflow y haz clic en Refresh para ver que el clúster esté completo.

Cuando el proceso run_dataproc_hadoop esté completo, ve al menú de navegación > Cloud Storage > Buckets y haz clic en el nombre de tu bucket para ver los resultados del conteo de palabras en la carpeta wordcount.

7. Cuando se completen todos los pasos en el DAG, cada uno tendrá un borde verde oscuro. Además, se borrará el clúster de Dataproc que creó.

¡Felicitaciones!

Ejecutaste correctamente un flujo de trabajo de Cloud Composer.

Próximos pasos

• Mira la presentación sobre Cloud Composer en NEXT 18 en San Francisco: Flexible, Easy Data Pipelines on Google Cloud with Cloud Composer (Cloud Next '18)
• Para ver el valor de una variable, ejecuta el subcomando variables de la CLI de Airflow con el argumento get o usa la interfaz web de Airflow.
• Para obtener información sobre la interfaz web de Airflow, consulta Accede a la interfaz web de Airflow.

Finalice su lab

Cuando haya completado el lab, haga clic en Finalizar lab. Google Cloud Skills Boost quitará los recursos que usó y limpiará la cuenta.

Tendrá la oportunidad de calificar su experiencia en el lab. Seleccione la cantidad de estrellas que corresponda, ingrese un comentario y haga clic en Enviar.

La cantidad de estrellas indica lo siguiente:

• 1 estrella = Muy insatisfecho
• 2 estrellas = Insatisfecho
• 3 estrellas = Neutral
• 4 estrellas = Satisfecho
• 5 estrellas = Muy satisfecho

Puede cerrar el cuadro de diálogo si no desea proporcionar comentarios.

Para enviar comentarios, sugerencias o correcciones, use la pestaña Asistencia.

Copyright 2020 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.