Configura una canalización de CI/CD para tu flujo de trabajo de procesamiento de datos

Last reviewed 2023-04-20 UTC

En este instructivo, se describe cómo configurar una canalización de integración continua e implementación continua (CI/CD) para procesar datos mediante la aplicación de métodos de CI/CD con productos administrados en Google Cloud. Los científicos y los analistas de datos pueden adaptar las metodologías de las prácticas de CI/CD para garantizar la alta calidad, el mantenimiento y la adaptabilidad de los procesos de datos y los flujos de trabajo. A continuación, se incluyen los métodos que puedes aplicar:

Control de versión del código fuente
Compilación, prueba y, también, implementación automática de apps
Aislamiento del entorno y separación de la producción
Procedimientos replicables para la configuración del entorno.

Este instructivo está dirigido a científicos y analistas de datos que compilan trabajos de procesamiento de datos en ejecución recurrentes para ayudar a estructurar la investigación y el desarrollo (I + D) a fin de mantener las cargas de trabajo de procesamiento de datos de forma sistemática y automática.

Arquitectura de implementación

En esta guía, usarás los siguientes productos de Google Cloud:

Usarás Cloud Build a fin de crear una canalización de CI/CD para compilar, implementar y probar un flujo de trabajo de procesamiento de datos y el procesamiento de datos en sí. Cloud Build es un servicio administrado que ejecuta la compilación en Google Cloud. Una compilación es una serie de pasos de compilación en la que cada paso se ejecuta en un contenedor de Docker.
Usarás Cloud Composer para definir y ejecutar los pasos del flujo de trabajo, como el inicio del procesamiento de datos, la prueba y la verificación de los resultados. Cloud Composer es un servicio administrado de Apache Airflow que ofrece un entorno en el que puedes crear, programar, supervisar y administrar flujos de trabajo complejos, como el flujo de trabajo de procesamiento de datos de este instructivo.
Usarás Dataflow para ejecutar el ejemplo WordCount de Apache Beam como un proceso de datos de muestra.

La canalización de CI/CD

En un nivel alto, la canalización de IC/EC consta de los pasos siguientes:

Cloud Build empaqueta la muestra de WordCount en un Archivo Java (JAR) de ejecución automática con el compilador de Maven. El compilador de Maven es un contenedor con Maven instalado. Cuando se configura un paso de compilación para usar el compilador de Maven, este ejecuta las tareas.
Cloud Build sube el archivo JAR a Cloud Storage.
Cloud Build ejecuta pruebas de unidades en el código del flujo de trabajo de procesamiento de datos y, luego, implementa el código del flujo de trabajo en Cloud Composer.
Cloud Composer reconoce el archivo JAR y ejecuta el trabajo de procesamiento de datos en Dataflow.

En el siguiente diagrama, se muestra una vista detallada de los pasos de la canalización de CI/CD.

Diagrama de la arquitectura de la canalización de CI/CD

En este instructivo, las implementaciones en los entornos de prueba y producción se separan en dos canalizaciones de Cloud Build diferentes: una de prueba y una de producción.

En el diagrama anterior, la canalización de prueba consta de los siguientes pasos:

Un desarrollador confirma los cambios de código en Cloud Source Repositories.
Los cambios de código activan una compilación de prueba en Cloud Build.
Cloud Build compila el archivo JAR de ejecución automática y lo implementa en el bucket de JAR de prueba en Cloud Storage.
Cloud Build implementa los archivos de prueba en los depósitos de archivos de prueba en Cloud Storage.
Cloud Build establece la variable en Cloud Composer para hacer referencia al archivo JAR recién implementado.
Cloud Build prueba el grafo acíclico dirigido (DAG) del flujo de trabajo de procesamiento de datos y lo implementa en el depósito de Cloud Composer en Cloud Storage.
El archivo de DAG del flujo de trabajo se implementa en Cloud Composer.
Cloud Build activa el flujo de trabajo de procesamiento de datos recién implementado para su ejecución.
Cuando se haya pasado la prueba de integración del flujo de trabajo de procesamiento de datos, se publicará un mensaje en Pub/Sub que contendrá una referencia al último JAR de ejecución automática (obtenido de las variables de Airflow) en el campo de datos del mensaje.

En el diagrama anterior, la canalización de producción consta de los siguientes pasos:

La canalización de implementación de producción se activa cuando se publica un mensaje en un tema de Pub/Sub.
Un desarrollador aprueba de forma manual la canalización de implementación de producción y se ejecuta la compilación.
Cloud Build copia el último archivo JAR de ejecución automática del bucket de JAR de prueba en el bucket de JAR de producción en Cloud Storage.
Cloud Build prueba el DAG del flujo de trabajo de procesamiento de datos de producción y lo implementa en el bucket de Cloud Composer en Cloud Storage.
El archivo de DAG del flujo de trabajo de producción se implementa en Cloud Composer.

En este instructivo, el flujo de trabajo de procesamiento de datos de producción se implementa en el mismo entorno de Cloud Composer que el flujo de trabajo de prueba para brindar una vista unificada de todos los flujos de trabajo de procesamiento de datos. A los efectos de este instructivo, los entornos se separan mediante diferentes depósitos de Cloud Storage para contener los datos de entrada y salida.

Para separar por completo los entornos, necesitas varios entornos de Cloud Composer creados en proyectos diferentes, que están separados de forma predeterminada. Esta separación ayuda a proteger el entorno de producción. Este método no se analiza en este instructivo. Para obtener más información sobre cómo acceder a los recursos en varios proyectos de Google Cloud, consulta Configura los permisos de la cuenta de servicio.

El flujo de trabajo de procesamiento de datos

Las instrucciones que indican cómo Cloud Composer ejecuta el flujo de trabajo de procesamiento de datos se definen en un grafo acíclico dirigido (DAG) escrito en Python. En el DAG, todos los pasos del flujo de trabajo de procesamiento de datos se definen junto con las dependencias entre ellos.

La canalización de CI/CD implementa de forma automática la definición de DAG de Cloud Source Repositories en Cloud Composer en cada compilación. Este proceso garantiza que Cloud Composer siempre esté actualizado con la última definición del flujo de trabajo sin necesidad de intervención humana.

En la definición de DAG para el entorno de prueba, se define un paso de prueba de extremo a extremo, además del flujo de trabajo de procesamiento de datos. El paso de prueba ayuda a garantizar que el flujo de trabajo de procesamiento de datos se ejecute de forma correcta.

El flujo de trabajo de procesamiento de datos se ilustra en el siguiente diagrama.

El flujo de trabajo de procesamiento de datos de cuatro pasos.

El flujo de trabajo de procesamiento de datos consta de los pasos siguientes:

Ejecuta el proceso de datos de WordCount en Dataflow.
Descarga los archivos de salida del proceso WordCount. El proceso WordCount genera tres archivos:
- download_result_1
- download_result_2
- download_result_3
Descarga el archivo de referencia, llamado download_ref_string.
Verifica el resultado con el archivo de referencia. Esta prueba de integración agrupa los tres resultados y compara todos los resultados con el archivo de referencia.
Publica un mensaje en Pub/Sub después de que finalice la prueba de integración.

El uso de un framework de organización de tareas como Cloud Composer para administrar el flujo de trabajo de procesamiento de datos ayuda a disminuir la complejidad del código del flujo de trabajo.

Las pruebas

Además de la prueba de integración que verifica el flujo de trabajo de procesamiento de datos de extremo a extremo, hay dos pruebas de unidades en este instructivo. Las pruebas de unidades son pruebas automáticas en el código de procesamiento de datos y el código del flujo de trabajo de procesamiento de datos. La prueba en el código de procesamiento de datos se escribe en Java y se ejecuta de forma automática durante el proceso de compilación de Maven. La prueba en el código del flujo de trabajo de procesamiento de datos se escribe en Python y se ejecuta como un paso de compilación independiente.

Objetivos

Configura el entorno de Cloud Composer.
Crea depósitos de Cloud Storage para los datos.
Crea las canalizaciones de compilación, prueba y producción.
Configura el activador de compilación.

Costos

En este documento, usarás los siguientes componentes facturables de Google Cloud:

Para generar una estimación de costos en función del uso previsto, usa la calculadora de precios. Es posible que los usuarios nuevos de Google Cloud califiquen para obtener una prueba gratuita.

Cuando finalices las tareas que se describen en este documento, puedes borrar los recursos que creaste para evitar que continúe la facturación. Para obtener más información, consulta Cómo realizar una limpieza.

Antes de comenzar

Accede a tu cuenta de Google Cloud. Si eres nuevo en Google Cloud, crea una cuenta para evaluar el rendimiento de nuestros productos en situaciones reales. Los clientes nuevos también obtienen $300 en créditos gratuitos para ejecutar, probar y, además, implementar cargas de trabajo.

En la página del selector de proyectos de la consola de Google Cloud, selecciona o crea un proyecto de Google Cloud.

Ir al selector de proyectos

Asegúrate de que la facturación esté habilitada para tu proyecto de Google Cloud.

Habilita las API de Cloud Build, Cloud Source Repositories, Cloud Composer, and Dataflow.

Habilita las API

En la página del selector de proyectos de la consola de Google Cloud, selecciona o crea un proyecto de Google Cloud.

Ir al selector de proyectos

Asegúrate de que la facturación esté habilitada para tu proyecto de Google Cloud.

Habilita las API de Cloud Build, Cloud Source Repositories, Cloud Composer, and Dataflow.

Habilita las API

Código de muestra

El código de muestra está en dos carpetas:

La carpeta env-setup contiene secuencias de comandos de shell para la configuración inicial del entorno de Google Cloud.
La carpeta source-code contiene código que se desarrolla a lo largo del tiempo, debe controlarse mediante el código fuente y activa procesos automáticos de compilación y prueba. Esta carpeta contiene las siguientes subcarpetas:
- La carpeta data-processing-code contiene el código fuente del proceso de Apache Beam.
- La carpeta workflow-dag contiene las definiciones de DAG del compositor destinadas a los flujos de trabajo de procesamiento de datos con los pasos para diseñar, implementar y probar el proceso de Dataflow.
- La carpeta build-pipeline contiene dos opciones de configuración de Cloud Build: una destinada a la canalización de prueba y otra para la canalización de producción. Esta carpeta también contiene una secuencia de comandos de compatibilidad para las canalizaciones.

A los fines de este instructivo, los archivos de código fuente para el procesamiento de datos y el flujo de trabajo de DAG se encuentran en diferentes carpetas en el mismo repositorio de código fuente. En un entorno de producción, los archivos de código fuente suelen estar en sus propios repositorios de código fuente y los administran equipos diferentes.

Configura tu entorno

En este instructivo, ejecutarás todos los comandos en Cloud Shell. Cloud Shell aparece como una ventana en la parte inferior de la consola de Google Cloud.

En la consola de Google Cloud, abre Cloud Shell.

Abra Cloud Shell

Clona el repositorio de código de muestra:

git clone https://github.com/GoogleCloudPlatform/ci-cd-for-data-processing-workflow.git

Ejecuta una secuencia de comandos para establecer variables de entorno:
```
cd ~/ci-cd-for-data-processing-workflow/env-setup
source set_env.sh
```
La secuencia de comandos establece las siguientes variables de entorno:
- El ID del proyecto de Google Cloud
- La región y la zona
- El nombre de los depósitos de Cloud Storage usados por la canalización de compilación y el flujo de trabajo de procesamiento de datos
Debido a que las variables de entorno no se conservan entre sesiones, si la sesión de Cloud Shell se cierra o se desconecta mientras se trabaja en el instructivo, deberás restablecer las variables de entorno.

Cree el entorno de Cloud Composer

En este instructivo, configurarás un entorno de Cloud Composer.

En Cloud Shell, agrega el rol Extensión del agente de servicio de la versión 2 de la API de Cloud Composer (roles/composer.ServiceAgentV2Ext) a la cuenta del agente de servicio de Cloud Composer:

gcloud projects add-iam-policy-binding $GCP_PROJECT_ID \
    --member serviceAccount:service-$PROJECT_NUMBER@cloudcomposer-accounts.iam.gserviceaccount.com \
    --role roles/composer.ServiceAgentV2Ext

En Cloud Shell, crea el entorno de Cloud Composer:
```
gcloud composer environments create $COMPOSER_ENV_NAME \
    --location $COMPOSER_REGION \
    --image-version composer-2.0.14-airflow-2.2.5
```
Nota: Por lo general, el aprovisionamiento del entorno de Cloud Composer toma alrededor de 15 minutos, pero puede demorar hasta una hora. Espera hasta que se complete este proceso antes de continuar con los siguientes pasos.
Ejecuta una secuencia de comandos para configurar las variables en el entorno de Cloud Composer. Las variables se necesitan en los DAG de procesamiento de datos.
```
cd ~/ci-cd-for-data-processing-workflow/env-setup
chmod +x set_composer_variables.sh
./set_composer_variables.sh
```
La secuencia de comandos establece las siguientes variables de entorno:
- El ID del proyecto de Google Cloud
- La región y la zona
- El nombre de los buckets de Cloud Storage que usan la canalización de compilación y el flujo de trabajo de procesamiento de datos

Extrae las propiedades del entorno de Cloud Composer

Cloud Composer usa un bucket de Cloud Storage para almacenar los DAG. Si mueves un archivo de definición de DAG al bucket, se activará Cloud Composer para leer los archivos de forma automática. Creaste el bucket de Cloud Storage para Cloud Composer cuando creaste el entorno de Cloud Composer. En el procedimiento siguiente, debes extraer la URL de los bucket s y, luego, configurar la canalización de CI/CD para implementar definiciones de DAG en el bucket de Cloud Storage de forma automática.

En Cloud Shell, exporta la URL del bucket como una variable de entorno:

export COMPOSER_DAG_BUCKET=$(gcloud composer environments describe $COMPOSER_ENV_NAME \
    --location $COMPOSER_REGION \
    --format="get(config.dagGcsPrefix)")

Exporta el nombre de la cuenta de servicio que usa Cloud Composer para tener acceso a los depósitos de Cloud Storage:

export COMPOSER_SERVICE_ACCOUNT=$(gcloud composer environments describe $COMPOSER_ENV_NAME \
    --location $COMPOSER_REGION \
    --format="get(config.nodeConfig.serviceAccount)")

Crea los depósitos de Cloud Storage

En esta sección, crearás un conjunto de depósitos de Cloud Storage para almacenar los siguientes elementos:

Los artefactos de los pasos intermedios del proceso de compilación
Los archivos de entrada y salida para el flujo de trabajo de procesamiento de datos
La ubicación de etapa de pruebas para que los trabajos de Dataflow almacenen los archivos binarios

Para crear los depósitos de Cloud Storage, completa el siguiente paso:

En Cloud Shell, cree buckets de Cloud Storage y otorgue permiso a la cuenta de servicio de Cloud Composer para ejecutar los flujos de trabajo de procesamiento de datos:
```
cd ~/ci-cd-for-data-processing-workflow/env-setup
chmod +x create_buckets.sh
./create_buckets.sh
```

Crea el tema de Pub/Sub

En esta sección, crearás un tema de Pub/Sub para recibir los mensajes enviados desde la prueba de integración del flujo de trabajo de procesamiento de datos a fin de activar automáticamente la canalización de compilación de producción.

En la consola de Google Cloud, ve a la página Temas de Pub/Sub.

Ir a la página Temas
Haz clic en Crear tema.
Para configurar el tema, completa los siguientes pasos:
- En ID del tema, ingresa integration-test-complete-topic.
- Confirma que la opción Agregar una suscripción predeterminada esté marcada.
- Deja las opciones restantes sin marcar.
- En Encriptación, selecciona Clave de encriptación administrada por Google.
- Haz clic en Crear tema.

Envíe el código fuente a Cloud Source Repositories

En este instructivo, tienes una base de código fuente que debes poner en el control de versiones. En el siguiente paso, se muestra cómo se desarrolla y se cambia una base de código con el tiempo. Cada vez que se envían cambios al repositorio, se activa la canalización para compilar, implementar y probar.

En Cloud Shell, envía la carpeta source-code a Cloud Source Repositories:

gcloud source repos create $SOURCE_CODE_REPO
cp -r ~/ci-cd-for-data-processing-workflow/source-code ~/$SOURCE_CODE_REPO
cd ~/$SOURCE_CODE_REPO
git init
git remote add google \
    https://source.developers.google.com/p/$GCP_PROJECT_ID/r/$SOURCE_CODE_REPO
git add .
git commit -m 'initial commit'
git push google master

Estos son comandos estándar para inicializar Git en un directorio nuevo y enviar el contenido a un repositorio remoto.

Crea canalizaciones de Cloud Build

En esta sección, crearás las canalizaciones de compilación con las que se compila, implementa y prueba el flujo de trabajo de procesamiento de datos.

Otorga acceso a la cuenta de servicio de Cloud Build

Cloud Build implementa los DAG de Cloud Composer y activa flujos de trabajo, que se habilitan cuando agregas acceso adicional a la cuenta de servicio de Cloud Build. Para obtener más información sobre las diferentes funciones disponibles cuando se trabaja con Cloud Composer, consulta la documentación de control de acceso.

En Cloud Shell, agrega la función composer.admin a la cuenta de servicio de Cloud Build para que el trabajo de Cloud Build pueda configurar variables de Airflow en Cloud Composer:
```
gcloud projects add-iam-policy-binding $GCP_PROJECT_ID \
    --member=serviceAccount:$PROJECT_NUMBER@cloudbuild.gserviceaccount.com \
    --role=roles/composer.admin
```
Agrega la función composer.worker a la cuenta de servicio de Cloud Build para que el trabajo de Cloud Build pueda activar el flujo de trabajo de datos en Cloud Composer:
```
gcloud projects add-iam-policy-binding $GCP_PROJECT_ID \
    --member=serviceAccount:$PROJECT_NUMBER@cloudbuild.gserviceaccount.com \
    --role=roles/composer.worker
```

Crea la canalización de compilación y prueba

Los pasos de canalización de compilación y prueba se configuran en el archivo de configuración YAML. En este instructivo, usarás imágenes de compiladores ya compiladas para git, maven, gsutil y gcloud a fin de ejecutar las tareas en cada paso de compilación. Usa las sustituciones de la variable de configuración para definir la configuración del entorno en el momento de la compilación. La ubicación del repositorio de código fuente se define mediante sustituciones de variables, al igual que las ubicaciones de los depósitos de Cloud Storage. La compilación necesita esta información para implementar el archivo JAR, los archivos de prueba y la definición de DAG.

En Cloud Shell, envía el archivo de configuración de canalización de compilación para crear la canalización en Cloud Build:
```
cd ~/ci-cd-for-data-processing-workflow/source-code/build-pipeline
gcloud builds submit --config=build_deploy_test.yaml --substitutions=\
REPO_NAME=$SOURCE_CODE_REPO,\
_DATAFLOW_JAR_BUCKET=$DATAFLOW_JAR_BUCKET_TEST,\
_COMPOSER_INPUT_BUCKET=$INPUT_BUCKET_TEST,\
_COMPOSER_REF_BUCKET=$REF_BUCKET_TEST,\
_COMPOSER_DAG_BUCKET=$COMPOSER_DAG_BUCKET,\
_COMPOSER_ENV_NAME=$COMPOSER_ENV_NAME,\
_COMPOSER_REGION=$COMPOSER_REGION,\
_COMPOSER_DAG_NAME_TEST=$COMPOSER_DAG_NAME_TEST
```
Este comando le indica a Cloud Build que ejecute una compilación con los siguientes pasos:
1. Compilación e implementación del archivo JAR de ejecución automática de WordCount
  1. Consultar el código fuente
  2. Compilar el código fuente de WordCount Beam en un archivo JAR de ejecución automática
  3. Almacenar el archivo JAR en Cloud Storage, donde Cloud Composer lo puede captar para ejecutar el trabajo de procesamiento de WordCount
2. Implementación y configuración del flujo de trabajo de procesamiento de datos en Cloud Composer
  1. Ejecutar la prueba de unidades en el código de operador personalizado que usa el DAG del flujo de trabajo
  2. Implementar el archivo de entrada de prueba y el archivo de referencia de prueba en Cloud Storage. El archivo de entrada de prueba es la entrada para el trabajo de procesamiento de WordCount. El archivo de referencia de prueba se usa como referencia para verificar el resultado del trabajo de procesamiento de WordCount
  3. Configurar las variables de Cloud Composer para que apunten al archivo JAR recién compilado
  4. Implementar la definición de DAG del flujo de trabajo en el entorno de Cloud Composer
3. Ejecutar el flujo de trabajo de procesamiento de datos en el entorno de prueba para activar el flujo de trabajo de procesamiento de pruebas

Verifica la compilación y la canalización de prueba

Después de enviar el archivo de compilación, verifica los pasos de compilación.

En la consola de Google Cloud, ve a la página Historial de compilaciones para ver una lista de todas las compilaciones anteriores y en ejecución.

Ir a la página Historial de compilación
Haz clic en la compilación que ahora esté en ejecución.
En la página Detalles de compilación (Build Steps), verifica que los pasos de compilación coincidan con los pasos descritos antes.

En la página Detalles de compilación, el campo Estado de la compilación indica Build successful cuando esta finaliza.

En Cloud Shell, verifica que el archivo JAR de muestra de WordCount se haya copiado en el bucket correcto:

gsutil ls gs://$DATAFLOW_JAR_BUCKET_TEST/dataflow_deployment*.jar

El resultado es similar a este:

gs://…-composer-dataflow-source-test/dataflow_deployment_e88be61e-50a6-4aa0-beac-38d75871757e.jar

Obtén la URL de tu interfaz web de Cloud Composer. Toma nota de la URL, porque la usarás en el siguiente paso.

gcloud composer environments describe $COMPOSER_ENV_NAME \
    --location $COMPOSER_REGION \
    --format="get(config.airflowUri)"

Use la URL del paso anterior para ir a la IU de Cloud Composer a fin de verificar la ejecución exitosa del DAG. Si en la columna Runs no se muestra información, espere unos minutos y vuelva a cargar la página.
1. Para verificar que el test_word_count del DAG del flujo de trabajo de procesamiento de datos se implemente y esté en modo de ejecución, mantén el cursor sobre el círculo verde claro bajo Runs y verifica que indique En ejecución.
2. Para ver el flujo de trabajo de procesamiento de datos en ejecución como un gráfico, haz clic en el círculo verde claro y, luego, en la página DAG Runs, haz clic en ID de DAG: test_word_count.
3. Vuelve a cargar la página Vista de gráfico para actualizar el estado de la ejecución de DAG actual. Por lo general, el flujo de trabajo toma entre tres y cinco minutos en completarse. Para verificar que la ejecución de DAG finalice de manera correcta, mantenga el cursor sobre cada tarea a fin de verificar que la información sobre la herramienta indique Estado: finalizado. La segunda y última tarea, llamada do_comparison, es la prueba de integración que verifica el resultado del proceso mediante su comparación con el archivo de referencia.
Una vez que se completa la prueba de integración, la última tarea, llamada publish_test_complete, publica un mensaje en el tema integration-test-complete-topic de Pub/Sub que se usará para activar la canalización de compilación de producción.
1. Para verificar que el mensaje publicado contenga la referencia correcta al último archivo JAR, podemos extraer el mensaje de la suscripción predeterminada de Pub/Sub integration-test-complete-topic-sub.
2. En la consola de Google Cloud, ve a la página Suscripciones.
  
  Ve a la página Suscripciones
3. Haz clic en integration-test-complete-topic-sub, selecciona la pestaña Mensaje y haz clic en Extraer
4. El resultado debería ser similar al siguiente ejemplo:

Crea la canalización de producción

Cuando el flujo de trabajo de procesamiento de pruebas se ejecute de forma correcta, podrás cambiar la versión actual del flujo de trabajo a producción. Existen varias formas de implementar el flujo de trabajo en producción:

De forma manual
De forma automática cuando todas las pruebas se aprueban en los entornos de prueba o de etapa de pruebas
De forma automática mediante un trabajo programado

En este instructivo, activarás automáticamente la compilación de producción cuando todas las pruebas se aprueben en el entorno de prueba. Para obtener más información sobre los enfoques automatizados, consulta Ingeniería de lanzamientos.

Antes de implementar el enfoque automatizado, verifica la compilación de implementación de producción mediante una implementación manual en producción. La compilación de la implementación de producción sigue estos pasos:

Se copia el archivo JAR de WordCount del bucket de prueba en el bucket de producción.
Se configuran las variables de Cloud Composer para que el flujo de trabajo de producción apunte al archivo JAR recién cambiado.
Se implementa la definición de DAG del flujo de trabajo de producción en el entorno de Cloud Composer y se ejecuta el flujo de trabajo.

Con las sustituciones de variables, se define el nombre del último archivo JAR que se implementa en producción con los depósitos de Cloud Storage que usa el flujo de trabajo de procesamiento de producción. Para crear la canalización de Cloud Build que implementa el flujo de trabajo de producción de Airflow, completa los siguientes pasos:

En Cloud Shell, imprime la variable de Cloud Composer para el nombre de archivo JAR a fin de poder leer el nombre del archivo JAR más reciente:

export DATAFLOW_JAR_FILE_LATEST=$(gcloud composer environments run $COMPOSER_ENV_NAME \
    --location $COMPOSER_REGION variables get -- \
    dataflow_jar_file_test 2>&1 | grep -i '.jar')

Usa el archivo de configuración de canalización de compilación deploy_prod.yaml, para crear la canalización en Cloud Build:

cd ~/ci-cd-for-data-processing-workflow/source-code/build-pipeline
gcloud builds submit --config=deploy_prod.yaml --substitutions=\
REPO_NAME=$SOURCE_CODE_REPO,\
_DATAFLOW_JAR_BUCKET_TEST=$DATAFLOW_JAR_BUCKET_TEST,\
_DATAFLOW_JAR_FILE_LATEST=$DATAFLOW_JAR_FILE_LATEST,\
_DATAFLOW_JAR_BUCKET_PROD=$DATAFLOW_JAR_BUCKET_PROD,\
_COMPOSER_INPUT_BUCKET=$INPUT_BUCKET_PROD,\
_COMPOSER_ENV_NAME=$COMPOSER_ENV_NAME,\
_COMPOSER_REGION=$COMPOSER_REGION,\
_COMPOSER_DAG_BUCKET=$COMPOSER_DAG_BUCKET,\
_COMPOSER_DAG_NAME_PROD=$COMPOSER_DAG_NAME_PROD

Verifica el flujo de trabajo de procesamiento de datos que creó la canalización de producción

Obtén la URL de la IU de Cloud Composer:

gcloud composer environments describe $COMPOSER_ENV_NAME \
    --location $COMPOSER_REGION \
    --format="get(config.airflowUri)"

Para verificar que el DAG del flujo de trabajo de procesamiento de datos de producción esté implementado, ve a la URL que recuperaste en el paso anterior y verifica que el DAG prod_word_count esté en la lista de DAG.
1. En la página DAGs, en la fila prod_word_count, haz clic en Activar DAG.
Haz clic en el logotipo de Airflow o vuelve a cargar la página para actualizar el estado de ejecución del DAG. Un círculo verde claro en la columna Ejecuciones indica que el DAG se encuentra en ejecución. Mantén el puntero sobre el círculo para ver la información sobre la herramienta que indica En ejecución.
Después de que la ejecución se realice con éxito, mantén el cursor sobre el círculo verde oscuro debajo de la columna Ejecuciones de DAG y verifica que indique Finalizado.

En Cloud Shell, enumera los archivos de resultados en el bucket de Cloud Storage:

gsutil ls gs://$RESULT_BUCKET_PROD

El resultado es similar a este:

gs://…-composer-result-prod/output-00000-of-00003
gs://…-composer-result-prod/output-00001-of-00003
gs://…-composer-result-prod/output-00002-of-00003

Crea activadores de Cloud Build

En esta sección, crearás los activadores de Cloud Build que vinculan los cambios del código fuente al proceso de compilación de prueba, y entre la canalización de prueba y la de compilación de producción.

Configura el activador de canalización de compilación de prueba

Configure un activador de Cloud Build que active una compilación nueva cuando los cambios se envíen a la rama principal del repositorio de código fuente.

En la consola de Google Cloud, ve a la página Activadores de compilación.

Ir a la página Activadores de compilación
Haga clic en Crear activador.
Para establecer la configuración del activador, siga estos pasos:
- En el campo Nombre, ingresa trigger-build-in-test-environment.
- En el menú desplegable Región, selecciona global (no regional).
- En Evento, haz clic en Enviar a una rama.
- En Fuente, selecciona data-pipeline-source.
- En el campo Nombre de la rama, ingresa master.
- En Configuración de compilación, haz clic en Archivo de configuración de Cloud Build (YAML o JSON).
- En Ubicación, haz clic en Repositorio.
- En el campo Ubicación del archivo de configuración de Cloud Build, ingresa build-pipeline/build_deploy_test.yaml.

En Cloud Shell, ejecuta el siguiente comando a fin de obtener todas las variables de sustitución necesarias para la compilación. Tome nota de estos valores, porque los necesitará en un paso posterior.

echo "_COMPOSER_DAG_BUCKET : ${COMPOSER_DAG_BUCKET}
_COMPOSER_DAG_NAME_TEST : ${COMPOSER_DAG_NAME_TEST}
_COMPOSER_ENV_NAME : ${COMPOSER_ENV_NAME}
_COMPOSER_INPUT_BUCKET : ${INPUT_BUCKET_TEST}
_COMPOSER_REF_BUCKET : ${REF_BUCKET_TEST}
_COMPOSER_REGION : ${COMPOSER_REGION}
_DATAFLOW_JAR_BUCKET : ${DATAFLOW_JAR_BUCKET_TEST}"

Nota: Se usa el par nombre-valor de salida para el siguiente paso

En la página Configuración del activador, en Variables de sustitución avanzadas, reemplaza las variables por los valores del entorno que obtuviste en el paso anterior. Agrega lo siguiente uno por vez y haz clic en + Agregar elemento (+ Add item) para cada par nombre-valor.
- _COMPOSER_DAG_BUCKET
- _COMPOSER_DAG_NAME_TEST
- _COMPOSER_ENV_NAME
- _COMPOSER_INPUT_BUCKET
- _COMPOSER_REF_BUCKET
- _COMPOSER_REGION
- _DATAFLOW_JAR_BUCKET
Haga clic en Crear.

Configura el activador de canalización de compilación de producción

Configura un activador de Cloud Build que active una compilación de producción cuando las pruebas se aprueben en el entorno de prueba y se publique un mensaje en el tema tests-complete de Pub/Sub. Este activador incluye un paso de aprobación en el que la compilación debe aprobarse de forma manual antes de que se ejecute la canalización de producción.

En la consola de Google Cloud, ve a la página Activadores de compilación.

Ir a la página Activadores de compilación
Haga clic en Crear activador.
Para establecer la configuración del activador, siga estos pasos:
- En el campo Nombre, ingresa trigger-build-in-prod-environment.
- En el menú desplegable Región, selecciona global (no regional).
- En Evento, haz clic en Mensaje de Pub/Sub.
- En Suscripción, selecciona integration-test-complete-topic.
- En Fuente, selecciona data-pipeline-source.
- En Revisión, selecciona Rama.
- En el campo Nombre de la rama, ingresa master.
- En Configuración de compilación, haz clic en Archivo de configuración de Cloud Build (YAML o JSON).
- En Ubicación, haz clic en Repositorio.
- En el campo Ubicación del archivo de configuración de Cloud Build, ingresa build-pipeline/deploy_prod.yaml.

echo "_COMPOSER_DAG_BUCKET : ${COMPOSER_DAG_BUCKET}
_COMPOSER_DAG_NAME_PROD : ${COMPOSER_DAG_NAME_PROD}
_COMPOSER_ENV_NAME : ${COMPOSER_ENV_NAME}
_COMPOSER_INPUT_BUCKET : ${INPUT_BUCKET_PROD}
_COMPOSER_REGION : ${COMPOSER_REGION}
_DATAFLOW_JAR_BUCKET_PROD : ${DATAFLOW_JAR_BUCKET_PROD}
_DATAFLOW_JAR_BUCKET_TEST : ${DATAFLOW_JAR_BUCKET_TEST}"

Nota: Se usa el par nombre-valor de salida para el siguiente paso

En la página Configuración del activador, en Variables de sustitución avanzadas, reemplaza las variables por los valores del entorno que obtuviste en el paso anterior. Agrega lo siguiente uno por vez y haz clic en + Agregar elemento (+ Add item) para cada par nombre-valor.
- _COMPOSER_DAG_BUCKET
- _COMPOSER_DAG_NAME_PROD
- _COMPOSER_ENV_NAME
- _COMPOSER_INPUT_BUCKET
- _COMPOSER_REGION
- _DATAFLOW_JAR_BUCKET_PROD
- _DATAFLOW_JAR_BUCKET_TEST
- _DATAFLOW_JAR_FILE_LATEST = $(body.message.data)
En Aprobación, marca Solicitar aprobación antes de que se ejecute la compilación.
Haga clic en Crear.

Prueba los activadores

Para probar el activador, agrega una palabra nueva al archivo de entrada de prueba y realiza el ajuste correspondiente en el archivo de referencia de prueba. Verifica que la canalización de compilación se active mediante un envío de confirmación a Cloud Source Repositories y que el flujo de trabajo de procesamiento de datos se ejecute de forma correcta con los archivos de prueba actualizados.

En Cloud Shell, agrega una palabra de prueba al final del archivo de prueba:

echo "testword" >>  ~/$SOURCE_CODE_REPO/workflow-dag/support-files/input.txt

Actualiza el archivo de referencia de resultados de prueba, ref.txt, para que coincida con los cambios realizados en el archivo de entrada de prueba:
```
echo "testword: 1" >>  ~/$SOURCE_CODE_REPO/workflow-dag/support-files/ref.txt
```

Confirme y envíe cambios a Cloud Source Repositories:

cd ~/$SOURCE_CODE_REPO
git add .
git commit -m 'change in test files'
git push google master

En la consola de Google Cloud, ve a la página Historial.

IR A LA PÁGINA HISTORIAL
Para verificar que una compilación de prueba nueva se active mediante el envío anterior a la rama principal, en la compilación que se está ejecutando actualmente, la columna Ref dice principal.

En Cloud Shell, obtenga la URL de la interfaz web de Cloud Composer:

gcloud composer environments describe $COMPOSER_ENV_NAME \
    --location $COMPOSER_REGION --format="get(config.airflowUri)"

Una vez finalizada la compilación, ve a la URL del comando anterior para verificar que el DAG de test_word_count se esté ejecutando.

Espera hasta que finalice la ejecución del DAG, que se indica cuando el círculo verde claro en la columna Ejecuciones de DAG desaparece. Por lo general, el proceso demora entre tres y cinco minutos.

En Cloud Shell, descarga los archivos de resultados de la prueba:

mkdir ~/result-download
cd ~/result-download
gsutil cp gs://$RESULT_BUCKET_TEST/output* .

Verifica que la palabra agregada hace poco esté en uno de los archivos de resultados:
```
grep testword output*
```
El resultado es similar a este:
```
output-00000-of-00003:testword: 1
```
En la consola de Google Cloud, ve a la página Historial.

Ir a la página Historial
Verifica que se haya activado una nueva compilación de producción cuando se complete la prueba de integración y que la compilación esté en espera de aprobación.
Marca la casilla junto a la compilación, haz clic en Aprobar y, luego, en Aprobar en el cuadro de confirmación para ejecutar la canalización de compilación de producción.
Una vez finalizada la compilación, ve a la URL del comando anterior y activa el DAG prod_word_count de forma manual para ejecutar la canalización de producción.

Realiza una limpieza

Para evitar que se apliquen cargos a tu cuenta de Google Cloud por los recursos usados en este instructivo, borra el proyecto que contiene los recursos o conserva el proyecto y borra los recursos individuales.

Borra el proyecto

En la consola de Google Cloud, ve a la página Administrar recursos.
Ir a Administrar recursos
En la lista de proyectos, elige el proyecto que quieres borrar y haz clic en Borrar.
En el diálogo, escribe el ID del proyecto y, luego, haz clic en Cerrar para borrar el proyecto.

Borra los recursos individuales

Si deseas mantener el proyecto que usaste en este instructivo, ejecuta los siguientes pasos a fin de borrar los recursos que creaste en este instructivo.

Para borrar el activador de Cloud Build, completa los siguientes pasos:
1. En la consola de Google Cloud, ve a la página Activadores.
  
  Ir a la página Activadores
2. Junto a los activadores que creaste, haz clic en Más y, luego, en Borrar.

En Cloud Shell, borra el entorno de Cloud Composer:

gcloud -q composer environments delete $COMPOSER_ENV_NAME \
    --location $COMPOSER_REGION

Borra los buckets de Cloud Storage y los archivos:

gsutil -m rm -r gs://$DATAFLOW_JAR_BUCKET_TEST \
    gs://$INPUT_BUCKET_TEST \
    gs://$REF_BUCKET_TEST \
    gs://$RESULT_BUCKET_TEST \
    gs://$DATAFLOW_STAGING_BUCKET_TEST \
    gs://$DATAFLOW_JAR_BUCKET_PROD \
    gs://$INPUT_BUCKET_PROD \
    gs://$RESULT_BUCKET_PROD \
    gs://$DATAFLOW_STAGING_BUCKET_PROD

Para borrar el tema y la suscripción predeterminada de Pub/Sub, ejecuta los siguientes comandos en Cloud Shell:

gcloud pubsub topics delete integration-test-complete-topic
gcloud pubsub subscriptions delete integration-test-complete-topic-sub

Borra el repositorio:

gcloud -q source repos delete $SOURCE_CODE_REPO

Borra los archivos y la carpeta que creaste:

rm -rf ~/ci-cd-for-data-processing-workflow
rm -rf ~/$SOURCE_CODE_REPO
rm -rf ~/result-download

¿Qué sigue?

Obtén más información sobre la entrega continua tipo GitOps con Cloud Build.
Consulta Automatiza la infraestructura con Cloud Composer.
Obtén más información sobre los Patrones comunes de casos prácticos de Dataflow.
Obtenga más información sobre la Ingeniería de lanzamientos.
Explora arquitecturas de referencia, diagramas y prácticas recomendadas sobre Google Cloud. Consulta nuestro Cloud Architecture Center.