Descripción general

En este lab, aprenderás a hacer lo siguiente:

• Implementar una canalización que tenga ramas
• Filtrar datos antes de escribirlos
• Agregar parámetros de línea de comandos personalizados a una canalización
• Convertir una canalización personalizada en una plantilla personalizada de Flex de Dataflow
• Ejecutar una plantilla de Flex de Dataflow

Requisitos previos:

• Conocimientos básicos sobre Python

En el lab anterior, creaste una canalización secuencial de extracción, transformación y carga básica, y utilizaste una plantilla de Dataflow equivalente para transferir el almacenamiento de datos por lotes en Google Cloud Storage. Esta canalización consta de una secuencia de transformaciones:

Sin embargo, muchas canalizaciones no mostrarán una estructura tan simple. En este lab, crearás una canalización no secuencial más sofisticada.

El caso de uso aquí es optimizar el consumo de recursos. Los productos varían según la forma en que consumen los recursos. Además, no todos los datos se utilizan de la misma manera dentro de una empresa. Algunos datos se consultarán con regularidad (por ejemplo, dentro de cargas de trabajo analíticas) y, otros, solo se usarán para la recuperación. En este lab, optimizarás la canalización del primer lab en pos del consumo de recursos. Para ello, almacenarás solo los datos que los analistas usarán en BigQuery y archivarás los demás datos en un servicio de almacenamiento muy duradero y de bajo costo, Coldline Storage en Google Cloud Storage.

Configuración y requisitos

Configuración de Qwiklabs

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.

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.

Configuración del entorno de desarrollo basado en notebooks de Jupyter

En este lab, ejecutarás todos los comandos en una terminal del notebook.

1. En el menú de navegación de la consola de Google Cloud, haz clic en Vertex AI > Workbench.

2. Haz clic en Habilitar API de Notebooks.

3. En la página de Workbench, selecciona NOTEBOOKS ADMINISTRADOS POR EL USUARIO y haz clic en CREAR NUEVO.

4. En el cuadro de diálogo Instancia nueva que se muestra, establece la región en y la zona en .

5. En Entorno, selecciona Apache Beam.

6. Haz clic en CREAR en la parte inferior del cuadro de diálogo.

Nota: El aprovisionamiento completo del entorno tarda de 3 a 5 minutos. Espera hasta que se complete este paso. Nota: Haz clic en Habilitar API de Notebooks para habilitarla.

7. Cuando el entorno esté listo, haz clic en el vínculo ABRIR JUPYTERLAB que se encuentra junto al nombre del notebook. Esto abrirá tu entorno en una nueva pestaña del navegador.

8. Luego, haz clic en Terminal. Esto abrirá una terminal en la que podrás ejecutar todos los comandos del lab.

Descarga el repositorio de código

A continuación, descargarás un repositorio de código que usarás en este lab.

1. En la terminal que acabas de abrir, ingresa lo siguiente:

git clone https://github.com/GoogleCloudPlatform/training-data-analyst cd /home/jupyter/training-data-analyst/quests/dataflow_python/

2. En el panel izquierdo de tu entorno de notebook, en el navegador de archivos, verás que se agregó el repo training-data-analyst.

3. Navega al repo clonado /training-data-analyst/quests/dataflow_python/. Verás una carpeta para cada lab. Cada una de ellas se divide en una subcarpeta lab con un código que debes completar y una subcarpeta solution con un ejemplo viable que puedes consultar como referencia si no sabes cómo continuar.

Nota: Para abrir un archivo y editarlo, simplemente debes navegar al archivo y hacer clic en él. Se abrirá el archivo, en el que puedes agregar o modificar código.

Haz clic en Revisar mi progreso para verificar el objetivo. Crear una instancia de notebook y clonar el repo del curso

Parte 1 del lab: Escribe canalizaciones con ramas

En esta parte del lab, escribirás una canalización con ramas que, a su vez, escribirá datos en Google Cloud Storage y en BigQuery.

Las transformaciones múltiples procesan la misma PCollection

Una forma de escribir una canalización con ramas es aplicar dos transformaciones diferentes a la misma PCollection, lo que da como resultado dos PCollections diferentes.

[PCollection1] = [Initial Input PCollection] | [A Transform] [PCollection2] = [Initial Input PCollection] | [A Different Transform]

Implementa una canalización con ramas

Si no puedes avanzar en esta sección o en secciones posteriores, la solución está disponible aquí.

Tarea 1: Agrega una rama para escribir en Cloud Storage

Para completar esta tarea, modifica una canalización existente agregando una rama que escriba en Cloud Storage.

Abre el lab adecuado

Si aún no lo has hecho, crea una terminal nueva en tu entorno de IDE y, luego, copia y pega el siguiente comando:

# Change directory into the lab cd 2_Branching_Pipelines/lab export BASE_DIR=$(pwd)

Configura el entorno virtual y las dependencias

Antes de comenzar a editar el código de la canalización en sí, debes asegurarte de haber instalado las dependencias necesarias.

Regresa a la terminal que abriste antes en tu IDE y, luego, crea un entorno virtual para nuestro trabajo en los siguientes labs:

sudo apt-get install -y python3-venv python3 -m venv df-env source df-env/bin/activate

A continuación, instala los paquetes que necesitaremos para ejecutar nuestra canalización:

python3 -m pip install -q --upgrade pip setuptools wheel python3 -m pip install apache-beam[gcp]

Por último, asegúrate de que la API de Dataflow esté habilitada:

gcloud services enable dataflow.googleapis.com

Configura el entorno de datos

# Create GCS buckets and BQ dataset cd $BASE_DIR/../.. source create_batch_sinks.sh # Generate event dataflow source generate_batch_events.sh # Change to the directory containing the practice version of the code cd $BASE_DIR

Abre my_pipeline.py en tu IDE, que se puede encontrar en 2_Branching_Pipelines/labs/. Desplázate hacia abajo hasta el método run(), en el que se define el cuerpo de la canalización. Actualmente, se ve de la siguiente manera:

(p | 'ReadFromGCS' >> beam.io.ReadFromText(input) | 'ParseJson' >> beam.Map(parse_json) | 'WriteToBQ' >> beam.io.WriteToBigQuery( output, schema=table_schema, create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED, write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE ) )

Modifica este código incorporando una transformación de ramificación nueva que escriba en Cloud Storage con textio.WriteToText antes de que cada elemento se convierta de json a dict.

Si no puedes avanzar en esta sección o en secciones posteriores, la solución está disponible aquí.

Haz clic en Revisar mi progreso para verificar el objetivo. Configura el entorno de datos

Tarea 2: Filtra datos por campo

Por el momento, la canalización nueva no consume menos recursos, puesto que todos los datos se almacenan dos veces. Para comenzar a mejorar el consumo de recursos, debemos reducir la cantidad de datos duplicados. El bucket de Google Cloud Storage está diseñado para funcionar como almacenamiento de archivos y copias de seguridad, por lo que es importante que todos los datos se guarden allí. Sin embargo, no es necesario que todos los datos se envíen a BigQuery.

Supongamos que los analistas de datos suelen observar a qué recursos acceden los usuarios en el sitio web y cómo esos patrones de acceso difieren en función de la geografía y el tiempo. Solo se necesitaría un subconjunto de los campos. Dado que analizamos los elementos json en diccionarios, podemos usar fácilmente el método pop para colocar un campo desde una función Python que admita llamadas:

def drop_field(element): element.pop('field_name') return element

A fin de completar esta tarea, usa una función Python que admita llamadas con beam.Map para descartar el campo user_agent, que nuestros analistas no usarán en BigQuery.

Tarea 3: Filtra datos por elemento

Existen muchas formas de filtrar en Apache Beam. Como estamos trabajando con una PCollection de diccionarios de Python, la forma más fácil será usar una función lambda (anónima) como nuestro filtro, una función que muestra un valor booleano, con beam.Filter. Por ejemplo:

purchases | beam.Filter(lambda element : element['cost_cents'] > 20*100)

Para completar esta tarea, agrega una transformación beam.Filter a la canalización. Puedes filtrar según los criterios que desees, pero como sugerencia intenta quitar las filas en las que num_bytes sea mayor o igual que 120.

Tarea 4: Agrega parámetros de línea de comandos personalizados

Actualmente, la canalización tiene una serie de parámetros hard-coded, incluida la ruta de acceso a la entrada y la ubicación de la tabla en BigQuery. Sin embargo, la canalización sería más útil si pudiera leer cualquier archivo JSON en Cloud Storage. Para agregar esta función, es necesario agregar elementos al conjunto de parámetros de la línea de comandos.

Actualmente, usamos un ArgumentParser para leer y analizar los argumentos de la línea de comandos. Luego, pasamos estos argumentos al objeto PipelineOptions() que especificamos cuando creamos nuestra canalización.

parser = argparse.ArgumentParser(description='...') # Define and parse arguments options = PipelineOptions() # Set options values from options p = beam.Pipeline(options=options)

Las PipelineOptions se usan para interpretar las opciones que lee ArgumentParser. Para agregar un argumento de línea de comandos nuevo al analizador, podemos usar la sintaxis:

parser.add_argument('--argument_name', required=True, help='Argument description')

Para acceder a un parámetro de la línea de comandos en el código, analiza los argumentos y consulta el campo en el diccionario resultante:

opts = parser.parse_args() arg_value = opts.arg_name

A fin de completar esta tarea, agrega parámetros de línea de comandos para la ruta de entrada, la ruta de salida de Google Cloud Storage y el nombre de la tabla de BigQuery, y actualiza el código de la canalización para acceder a esos parámetros en lugar de constantes.

Tarea 5: Agrega campos NULLABLE a tu canalización

Probablemente notaste que la tabla de BigQuery creada en el último lab tenía un esquema con todos los campos REQUIRED como el siguiente:

Es posible que quieras crear un esquema Apache Beam con campos NULLABLE en los que falten datos, tanto para la ejecución de la canalización, como para una tabla de BigQuery resultante con un esquema que refleje esto.

Podemos actualizar el esquema JSON de BigQuery si agregamos un nuevo mode de propiedad para el campo que queremos que sea nulo:

{ "name": "field_name", "type": "STRING", "mode": "NULLABLE" }

Para completar esta tarea, marca los campos lat y lon como anulables en el esquema de BigQuery.

Tarea 6: Ejecuta tu canalización desde la línea de comandos

Para completar esta tarea, ejecuta tu canalización desde la línea de comandos y pasa los parámetros adecuados. Recuerda tomar nota del esquema resultante de BigQuery para los campos NULLABLE. El código debería ser similar al siguiente:

# Set up environment variables export PROJECT_ID=$(gcloud config get-value project) export REGION={{{project_0.startup_script.lab_region|Region}}} export BUCKET=gs://${PROJECT_ID} export COLDLINE_BUCKET=${BUCKET}-coldline export PIPELINE_FOLDER=${BUCKET} export RUNNER=DataflowRunner export INPUT_PATH=${PIPELINE_FOLDER}/events.json export OUTPUT_PATH=${PIPELINE_FOLDER}-coldline/pipeline_output export TABLE_NAME=${PROJECT_ID}:logs.logs_filtered cd $BASE_DIR python3 my_pipeline.py \ --project=${PROJECT_ID} \ --region=${REGION} \ --stagingLocation=${PIPELINE_FOLDER}/staging \ --tempLocation=${PIPELINE_FOLDER}/temp \ --runner=${RUNNER} \ --inputPath=${INPUT_PATH} \ --outputPath=${OUTPUT_PATH} \ --tableName=${TABLE_NAME} Si tu canalización se crea correctamente, pero observas muchos errores debido a un código o a una configuración incorrecta en el servicio de Dataflow, puedes configurar RUNNER de nuevo en “DirectRunner” para ejecutarlo de forma local y recibir resultados más rápido. Este enfoque funciona en este caso porque el conjunto de datos es pequeño y no estás usando ninguna función que no sea compatible con DirectRunner.

Tarea 7: Verifica los resultados de la canalización

Navega a la página de trabajos de Cloud Dataflow y observa el trabajo mientras se ejecuta. Tu gráfico debería ser similar al siguiente:

Haz clic en el nodo que representa su función Filter que, en la imagen anterior, se llama FilterFn. En el panel que aparece en el lado derecho, deberías ver que se agregaron más elementos como entradas que los que se escribieron como salidas.

Ahora, haz clic en el nodo que representa la escritura en Cloud Storage. Dado que se escribieron todos los elementos, esta cantidad debe coincidir con la cantidad de elementos de la entrada de la función de filtro.

Una vez finalizada la canalización, consulta tu tabla para examinar los resultados en BigQuery. Ten en cuenta que la cantidad de registros en la tabla debe coincidir con la cantidad de elementos que generó la función Filtro.

Haz clic en Revisar mi progreso para verificar el objetivo. Ejecuta tu canalización desde la línea de comandos

Parte 2 del Lab: Plantillas personalizadas de Dataflow

Una canalización que acepta parámetros de la línea de comandos es mucho más útil que una que tiene esos parámetros en modalidad hard-coded. Sin embargo, para ejecutar la canalización, se debe crear un entorno de desarrollo. Si se espera que múltiples usuarios diferentes vuelvan a ejecutar la canalización, o que la canalización se vuelva a ejecutar en una variedad de contextos diferentes, una opción todavía mejor sería usar una plantilla de Dataflow.

Hay muchas plantillas de Dataflow que ya se crearon como parte de Google Cloud Platform (puedes explorarlas aquí). Sin embargo, ninguna realiza la misma función que la canalización de este lab. En su lugar, en esta parte del lab, convertirás la canalización en una plantilla personalizada de Flex de Dataflow más nueva (en lugar de una plantilla tradicional personalizada).

Convertir una canalización en una plantilla personalizada de Flex de Dataflow requiere usar un contenedor de Docker para empaquetar tu código y las dependencias, un Dockerfile a fin de describir qué código se compilará, Cloud Build para crear el contenedor subyacente que se ejecutará en el entorno de ejecución para crear el trabajo real y un archivo de metadatos para describir los parámetros del trabajo.

Tarea 1: Crea una imagen de contenedor de la plantilla personalizada de Flex de Dataflow

1. Primero, habilita el uso de caché de Kaniko de forma predeterminada. Kaniko almacena en caché los artefactos de compilación de contenedores, por lo que usar esta opción acelera las compilaciones posteriores. También usaremos pip3 freeze para registrar los paquetes y las versiones que se usan en nuestro entorno.

gcloud config set builds/use_kaniko True

2. A continuación, crearemos nuestro Dockerfile. Esto especificará el código y las dependencias que necesitamos usar.

   a. Para completar esta tarea, crea un archivo nuevo en la carpeta dataflow_python/2_Branching_Pipelines/lab en el explorador de archivos de tu IDE.

   b. Para crear un archivo nuevo, haz clic en Archivo >> Nuevo >> Archivo de texto.

   c. Cámbiale el nombre al archivo a Dockerfile. Para hacerlo, haz clic en él con el botón derecho.

   d. Abre Dockerfile en el panel de edición y haz clic en el archivo para abrirlo.

   e. Copia el siguiente código en el archivo Dockerfile y guárdalo:

FROM gcr.io/dataflow-templates-base/python3-template-launcher-base ARG WORKDIR=/dataflow/template RUN mkdir -p ${WORKDIR} WORKDIR ${WORKDIR} RUN apt-get update && apt-get install -y libffi-dev && rm -rf /var/lib/apt/lists/* COPY my_pipeline.py . ENV FLEX_TEMPLATE_PYTHON_PY_FILE="${WORKDIR}/my_pipeline.py" RUN python3 -m pip install apache-beam[gcp]==2.25.0

3. Por último, usa Cloud Build para compilar la imagen del contenedor.

export TEMPLATE_IMAGE="gcr.io/$PROJECT_ID/dataflow/my_pipeline:latest" gcloud builds submit --tag $TEMPLATE_IMAGE .

4. La compilación y el envío del contenedor tardará unos minutos.

Haz clic en Revisar mi progreso para verificar el objetivo. Crea una imagen de contenedor de una plantilla personalizada de Flex de Dataflow

Tarea 2: Crea la plantilla de Flex y almacénala en etapa intermedia

1. Para ejecutar una plantilla, debes crear un archivo de especificaciones de la plantilla en Cloud Storage que contenga toda la información necesaria para ejecutar el trabajo, como la información y los metadatos del SDK.

   a. Crea un archivo nuevo en la carpeta dataflow_python/2_Branching_Pipelines/lab del explorador de archivos de tu IDE.

   b. Para crear un Archivo nuevo, haz clic en Archivo >> Nuevo >> Archivo de texto.

   c. Cámbiale el nombre al archivo a metadata.json. Para hacerlo, haz clic en él con el botón derecho.

   d. Abre el archivo metadata.json en el panel de edición. Para abrir el archivo metadata.json, haz clic en él con el botón derecho y, luego, selecciona Abrir con >> Editor.

   e. Para completar esta tarea, crea un archivo metadata.json en el siguiente formato que considere todos los parámetros de entrada que espera tu canalización. Si es necesario, consulta la solución aquí. Para hacerlo, debes escribir tu propia comprobación de regex de parámetros. Si bien no es una práctica recomendada, ".*" coincidirá con cualquier entrada.

{ "name": "My Branching Pipeline", "description": "A branching pipeline that writes raw to GCS Coldline, and filtered data to BQ", "parameters": [ { "name": "inputPath", "label": "Input file path.", "helpText": "Path to events.json file.", "regexes": [ ".*\\.json" ] }, { "name": "outputPath", "label": "Output file location", "helpText": "GCS Coldline Bucket location for raw data", "regexes": [ "gs:\\/\\/[a-zA-z0-9\\-\\_\\/]+" ] }, { "name": "tableName", "label": "BigQuery output table", "helpText": "BigQuery table spec to write to, in the form 'project:dataset.table'.", "regexes": [ "[^:]+:[^.]+[.].+" ] } ] }

2. Luego, compila la plantilla real y almacénala en etapa intermedia:

export TEMPLATE_PATH="gs://${PROJECT_ID}/templates/mytemplate.json" # Will build and upload the template to GCS # You may need to opt-in to beta gcloud features gcloud beta dataflow flex-template build $TEMPLATE_PATH \ --image "$TEMPLATE_IMAGE" \ --sdk-language "PYTHON" \ --metadata-file "metadata.json"

3. Verifica que el archivo se haya subido a la ubicación de la plantilla en Cloud Storage.

Haz clic en Revisar mi progreso para verificar el objetivo. Crear la plantilla de Flex y almacenarla en etapa intermedia

Tarea 3: Ejecuta la plantilla desde la IU

Para completar la tarea, sigue estas instrucciones.

1. Ve a la página de Cloud Dataflow en la consola de Google Cloud.

2. Haz clic en CREAR UN TRABAJO A PARTIR DE UNA PLANTILLA.

3. Ingresa un nombre válido para el trabajo en el campo Nombre del trabajo.

4. Selecciona Plantilla personalizada en el menú desplegable de plantillas de Cloud Dataflow.

5. Ingresa la ruta de Cloud Storage a tu archivo de plantilla en el campo de la ruta de la plantilla de Cloud Storage.

6. Ingresa los elementos adecuados en Parámetros obligatorios. a. En Input file path, ingresa ,

   b. En Output file location, ingresa ,

   c. En BigQuery output table, ingresa .

7. Haz clic en Ejecutar trabajo.

Nota: No necesitas especificar un bucket de almacenamiento en etapa intermedia, ya que Dataflow creará uno privado en tu proyecto con el número de proyecto, similar a .

8. Si revisas la consola de Compute Engine, verás una VM de iniciador temporal que se creó para ejecutar el contenedor y poder iniciar tu canalización con los parámetros proporcionados.

Tarea 4: Ejecuta la plantilla con gcloud

1. Uno de los beneficios de usar las plantillas de Dataflow es que puedes ejecutarlas desde muchos contextos distintos del entorno de desarrollo. Para demostrarlo, usa gcloud a fin de ejecutar una plantilla de Dataflow desde la línea de comandos.

2. Para completar esta tarea, ejecuta el siguiente comando en tu terminal y modifica los parámetros según corresponda:

export PROJECT_ID=$(gcloud config get-value project) export REGION={{{project_0.startup_script.lab_region|Region}}} export JOB_NAME=mytemplate-$(date +%Y%m%H%M$S) export TEMPLATE_LOC=gs://${PROJECT_ID}/templates/mytemplate.json export INPUT_PATH=gs://${PROJECT_ID}/events.json export OUTPUT_PATH=gs://${PROJECT_ID}-coldline/template_output/ export BQ_TABLE=${PROJECT_ID}:logs.logs_filtered gcloud beta dataflow flex-template run ${JOB_NAME} \ --region=$REGION \ --template-file-gcs-location ${TEMPLATE_LOC} \ --parameters "inputPath=${INPUT_PATH},outputPath=${OUTPUT_PATH},tableName=${BQ_TABLE}"

3. Asegúrate de que tu canalización se complete correctamente.

Haz clic en Revisar mi progreso para verificar el objetivo. Ejecuta la plantilla desde la IU y con gcloud

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 2026 Google LLC. Todos los derechos reservados. Google y el logotipo de Google son marcas de Google LLC. El resto de los nombres de productos y empresas pueden ser marcas de las respectivas empresas a las que están asociados.