Descripción general

Los flujos de trabajo son un tema recurrente en las estadísticas 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 este 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 conteo 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.

Active Cloud Shell

Cloud Shell es una máquina virtual que contiene herramientas de desarrollo y un directorio principal persistente de 5 GB. Se ejecuta en Google Cloud. Cloud Shell proporciona acceso de línea de comandos a sus recursos de Google Cloud. gcloud es la herramienta de línea de comandos de Google Cloud, la cual está preinstalada en Cloud Shell y es compatible con la función de autocompletado con tabulador.

1. En el panel de navegación de Google Cloud Console, haga clic en Activar Cloud Shell ().

2. Haga clic en Continuar.
   El aprovisionamiento y la conexión al entorno tardan solo unos momentos. Una vez que se conecte, también estará autenticado, y el proyecto estará configurado con su PROJECT_ID. Por ejemplo:

Comandos de muestra

• Si desea ver el nombre de cuenta activa, use este comando:

gcloud auth list

(Resultado)

Credentialed accounts: - <myaccount>@<mydomain>.com (active)

(Resultado de ejemplo)

Credentialed accounts: - google1623327_student@qwiklabs.net

• Si desea ver el ID del proyecto, use este comando:

gcloud config list project

(Resultado)

[core] project = <project_ID>

(Resultado de ejemplo)

[core] project = qwiklabs-gcp-44776a13dea667a6

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 el Menú de navegación () de la consola de Google Cloud, haga clic en IAM y administración > IAM.

2. Confirma que aparezca la cuenta de servicio predeterminada de Compute {project-number}-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.

Si no aparece la cuenta 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.

2. En la tarjeta Información del proyecto, copia el Número de proyecto.

3. En el menú de navegación, haz clic en IAM y administración > IAM.

4. En la parte superior de la página IAM, haga clic en Agregar.

5. En Principales nuevas, escriba lo siguiente:

{número-del-proyecto}-compute@developer.gserviceaccount.com

Reemplaza {project-number} por el número de tu proyecto.

6. En Seleccionar un rol, elige Básico (o Proyecto) > Editor.

7. Haz clic en Guardar.

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

Para garantizar el acceso a las APIs necesarias, reinicia la conexión a la API de Kubernetes Engine.

1. En la consola de Google Cloud, busca y registra el Número de proyecto en el cuadro Información del proyecto.

2. En la consola, ingresa API de Kubernetes Engine en la barra de búsqueda superior. Haz clic en el resultado de la API de Kubernetes Engine.

3. Haz clic en Administrar.

4. Haz clic en Inhabilitar API.

   • Si se te solicita confirmar, haz clic en Inhabilitar.

   • De nuevo, cuando aparezca el mensaje Do you want to disable Kubernetes Engine API and its dependent APIs?, haz clic en Inhabilitar.

5. Haz clic en Habilitar.

Cuando se haya habilitado de nuevo la API, se mostrará en la página la opción para inhabilitarla.

Tarea 2. Asegúrate de que la API de Cloud Composer esté habilitada correctamente

Reinicia la conexión a la API de Cloud Composer. En el paso anterior, el reinicio de la API de Kubernetes Engine provocó que se inhabilitara la API de Cloud Composer.

1. En la consola de Google Cloud, ingresa API de Cloud Composer en la barra de búsqueda superior. Haz clic en el resultado de la API de Cloud Composer.

2. Haz clic en Habilitar.

Cuando se haya habilitado de nuevo la API, se mostrará en la página la opción para inhabilitarla.

Tarea 3: 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. En el campo Buscar de la barra de título de la consola de Google Cloud, escribe cloud composer y, luego, haz clic en Composer.

2. Haz clic en Crear entorno y selecciona Composer 3.

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)

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

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

5. Haz clic en Crear.

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

El entorno puede tardar entre 10 y 20 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 se usará como 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 un nombre universalmente único al bucket 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, ya que lo usarás como una 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 4: 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 quieres ejecutar, organizadas para 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 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, …

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

Tarea 5: 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. * 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_sample_quickstart', # 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='n1-standard-2', worker_machine_type='n1-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 quickstart y se ejecuta una vez al día.

with models.DAG( 'composer_sample_quickstart', # 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 6: 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 7. Usa la IU de Airflow

Realiza lo siguiente para acceder a la interfaz web de Airflow por medio de la consola de Google Cloud:

1. Regrese 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 8: 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 y gce_zone:

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.

4. 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 9: 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:

gsutil 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 Configuración del entorno 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:

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

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

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.

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.

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 de Cloud Composer en NEXT 18 en San Francisco: https://www.youtube.com/watch?v=GeNFEtt-D4k.
• 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 su lab, haga clic en Finalizar lab. Qwiklabs quitará los recursos que usó y limpiará la cuenta por usted.

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.