Descripción general

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

• Escribir una canalización que agrupe el tráfico del sitio por usuario
• Escribir una canalización que agrupe el tráfico del sitio por minuto
• Implementar la renderización en ventanas en datos de series temporales

Requisitos previos

NOTA: El lab es de nivel avanzado y se requiere un conocimiento avanzado de Python.

Configuración y requisitos

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.

Parte A: Configuración del entorno de desarrollo de las instancias de Workbench

En este lab, ejecutarás todos los comandos en una terminal de tu notebook de instancia de Workbench.

1. En el menú de navegación () de la consola de Google Cloud, selecciona Vertex AI. También puedes ir al panel de Vertex AI.

2. Haz clic en Habilitar todas las APIs recomendadas. Ahora, verifiquemos que la API de Notebook esté habilitada.

3. En el menú de navegación, haz clic en Workbench.

   En la parte superior de la página de Workbench, asegúrate de estar en la vista Instancias.

4. Haz clic en Crear nueva.

5. Configura la instancia:

Nombre	Región	Zona	Opciones avanzadas (opcionales)
lab-workbench			Si es necesario, haz clic en "Opciones avanzadas" para realizar personalizaciones adicionales (p. ej., tipo de máquina, tamaño del disco).

6. Haz clic en Crear.

La instancia tardará algunos minutos en crearse. Se mostrará una marca de verificación verde junto a su nombre cuando esté lista.

7. Haz clic en ABRIR JUPYTERLAB junto al nombre de la instancia para iniciar la interfaz de JupyterLab. Se abrirá una pestaña nueva en el navegador.

8. Luego, haz clic en Terminal. Esto abrirá una terminal en la que podrá 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

Tarea 1: Agrupa el tráfico del sitio por usuario

En esta parte del lab, escribirás una canalización que realizará las siguientes acciones:

1. Leer el tráfico del día desde un archivo de Cloud Storage
2. Convertir cada evento en un objeto CommonLog
3. Sumar la cantidad de hits de cada usuario único, agrupando cada objeto por ID de usuario y combinando los valores para obtener la cantidad total de hits para ese usuario específico
4. Realizar agregaciones adicionales en cada usuario
5. Escribir los datos resultantes en BigQuery

Tarea 2: Genera datos sintéticos

Al igual que en los labs anteriores, el primer paso es generar datos para que la canalización los procese. Abre el entorno del lab y genera los datos como lo hiciste anteriormente:

Abre el lab adecuado

• En la terminal de tu IDE, ejecuta los siguientes comandos:

# Cambiar el directorio en el lab cd 3_Batch_Analytics/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.

1. Ejecuta el siguiente comando para crear un entorno virtual para tu trabajo en este lab:

sudo apt-get update && sudo apt-get install -y python3-venv # Crear y activar un entorno virtual python3 -m venv df-env source df-env/bin/activate

2. Luego, instala los paquetes que necesitarás para ejecutar tu canalización:

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

3. Asegúrate de que esté habilitada la API de Dataflow:

gcloud services enable dataflow.googleapis.com

Configura el entorno de datos

# Crear buckets de GCS y un conjunto de datos de BQ cd $BASE_DIR/../.. source create_batch_sinks.sh # Generar evento de Dataflow source generate_batch_events.sh # Cambiar al directorio que contiene la versión de práctica del código cd $BASE_DIR

La secuencia de comandos crea un archivo llamado events.json que contiene líneas similares a las siguientes:

{"user_id": "-6434255326544341291", "ip": "192.175.49.116", "timestamp": "2019-06-19T16:06:45.118306Z", "http_request": "\"GET eucharya.html HTTP/1.0\"", "lat": 37.751, "lng": -97.822, "http_response": 200, "user_agent": "Mozilla/5.0 (compatible; MSIE 7.0; Windows NT 5.01; Trident/5.1)", "num_bytes": 182}

Luego, copiará automáticamente este archivo en tu bucket de Google Cloud Storage en .

• Navega a Google Cloud Storage y confirma que tu bucket de almacenamiento contenga un archivo llamado events.json.

Haz clic en Revisar mi progreso para verificar el objetivo. Generar datos sintéticos

Tarea 3: Suma las vistas de páginas por usuario

Esta tarea tiene dos minidesafíos. Puedes consultar la solución

En el explorador de archivos, ve a la ruta mencionada a continuación y abre el archivo batch_user_traffic_pipeline.py.

training-data-analyst/quests/dataflow_python/3_Batch_Analytics/lab

Minidesafío #TODO 1:

Esta canalización ya contiene el código necesario para aceptar las opciones de la línea de comandos de la ruta de entrada y el nombre de la tabla de salida, así como el código para leer en los eventos de Google Cloud Storage, analizarlos y escribir resultados en BigQuery. Sin embargo, faltan algunas partes importantes que se marcaron como #TODOs.

Esta es una tarea de modelado de datos. En este paso, debes considerar, después de agrupar todos los eventos de registro por usuario, qué información quieres calcular para cada uno.

Debes definir la estructura (el esquema) que contendrá los resultados de tu agregación. Debes consultar la clase CommonLog para ver qué campos están disponibles para agregar.

Cómo resolverlo:

1. Identifica la clave: El nombre de la clase es PerUserAggregation, por lo que la información principal que debes conservar es el user_id.

2. Elige las métricas que se calcularán: ¿Qué puedes calcular a partir de la colección de entradas CommonLog de un usuario?

   • Un recuento: ¿Cuántas veces accedió el usuario al servidor?
   • Una suma: ¿Cuál es la cantidad total de bytes (num_bytes) que descargó?
   • Un mín./máx.: ¿Cuál fue la marca de tiempo de su primera y última actividad?

3. Por ejemplo:

user_id: str page_views: int ...

Minidesafío #TODO 2:

Este es un requisito técnico del framework de Apache Beam. El desafío consiste en poner a prueba tus conocimientos sobre cómo Beam maneja los tipos de datos personalizados.

Cuando Apache Beam ejecuta una canalización, a menudo necesita enviar datos entre diferentes computadoras (llamadas trabajadores). Para ello, debe serializar tu objeto de Python (como una instancia PerUserAggregation) en un flujo de bytes, enviarlo a través de la red y, luego, deserializarlo de nuevo en un objeto en el otro extremo. Un Coder es el objeto que le indica a Beam cómo realizar esta serialización y deserialización.

Si no le indicas a Beam cómo codificar o decodificar tu clase PerUserAggregation personalizada, la canalización fallará con un error.

Cómo resolverlo:

La solución está justo en la línea que se encuentra sobre las #TODOs. Beam proporciona un RowCoder que funciona perfectamente con las clases NamedTuple. Solo debes registrar un RowCoder para tu nueva clase PerUserAggregation, tal como se hizo para CommonLog.

4. Por ejemplo:

beam.coders.registry.register_coder(PerUserAggregation, ...)

Tarea 4: Ejecuta tu canalización

Regresa a la terminal y ejecuta el siguiente comando para ejecutar tu canalización con el servicio de Cloud Dataflow. Si tienes problemas, puedes ejecutarla con DirectRunner o consultar la solución.

• En el siguiente fragmento de código, reemplaza los campos ENTER_REGION_ID y ENTER_ZONE_ID según la tabla que aparece a continuación.

Región	Zona

• Reemplaza los valores regionales y zonales según las especificaciones de tu lab.

# 1. Establece las variables de entorno export PROJECT_ID=$(gcloud config get-value project) export REGION=ENTER_REGION_ID export ZONE=ENTER_ZONE_ID export BUCKET=gs://${PROJECT_ID} export PIPELINE_FOLDER=${BUCKET} export RUNNER=DataflowRunner export INPUT_PATH=${PIPELINE_FOLDER}/events.json export TABLE_NAME=${PROJECT_ID}:logs.user_traffic # 2. Verifica que exista el archivo de entrada # Este comando NO debería devolver un error. echo "Verifying input file exists at ${INPUT_PATH}..." gcloud storage ls ${INPUT_PATH} # 3. Ejecuta la secuencia de comandos de canalización echo "Running the Dataflow pipeline..." python3 batch_user_traffic_pipeline.py \ --project=${PROJECT_ID} \ --region=${REGION} \ --worker_zone=${ZONE} \ --staging_location=${PIPELINE_FOLDER}/staging \ --temp_location=${PIPELINE_FOLDER}/temp \ --runner=${RUNNER} \ --input_path=${INPUT_PATH} \ --table_name=${TABLE_NAME}

• Puedes verificar tu trabajo enviado en el Panel de Dataflow

• Una vez que el estado del trabajo sea correcto, verifiquemos los resultados en BigQuery. Para completar esta tarea, espera unos minutos hasta que se complete la canalización. Luego, navega a BigQuery y consulta la tabla user_traffic.

Haz clic en Revisar mi progreso para verificar el objetivo. Agregar tráfico del sitio por usuario y ejecutar la canalización

Parte B: Agrega el tráfico del sitio por minuto

En esta parte del lab, crearás una nueva canalización llamada batch_minute_traffic. Es una extensión de los principios básicos del análisis por lotes usado en batch_user_traffic, solo que, en lugar de hacer una agregación según los usuarios del lote completo, se hace una agregación según el momento en que ocurrieron los eventos. Hay varias #TODOs que debes corregir. También puedes obtener ayuda de la solución.

En el IDE, navega a la siguiente ruta y abre el archivo batch_minute_traffic_pipeline.py.

3_Batch_Analytics/lab Nota: Antes de empezar a trabajar con la secuencia de comandos principal (batch_minute_traffic_pipeline.py), asegúrate de revisar los dos archivos de ayuda pipeline_utils.py y setup.py. Para ejecutar nuestros trabajos de Dataflow, utilizaremos estos archivos auxiliares, ya que pipeline_utils contiene lógica personalizada de Python (clases y funciones) en un módulo independiente que se puede empaquetar y enviar a trabajadores remotos de Dataflow. Por otro lado, setup.py actúa como el manual de instrucciones para Dataflow, ya que le indica exactamente cómo agrupar tu archivo pipeline_utils.py en un paquete adecuado para instalar en cada trabajador.

Tarea 5: Agrega marcas de tiempo a cada elemento

Estos elementos #TODOs te guían en la creación de una canalización de agregación de series temporales clásica en Apache Beam. Cada uno representa un concepto central en el procesamiento por lotes y de transmisión. El objetivo de esta canalización es procesar un archivo JSON de eventos web (events.json), contar cuántos eventos ocurrieron cada minuto y escribir estos recuentos minuto a minuto en una tabla de BigQuery.

El flujo de la canalización es el siguiente: Read Text -> Parse to CommonLog -> TODOs -> Write to BigQuery

Las #TODOs son el concepto central de la canalización, en el que se produce la lógica de agregación real.

Minidesafío #TODO 1:

Tienes una colección de objetos CommonLog. El siguiente paso en la canalización es agrupar estos eventos por tiempo (WindowByMinute). Apache Beam no puede hacerlo, a menos que conozca el tiempo del evento para cada dato. Tu desafío es indicarle a Beam cómo encontrar esta marca de tiempo en tu objeto CommonLog.

Cómo resolverlo:

1. La función add_timestamp (definida en pipeline_utils.py) analiza la cadena de marca de tiempo de cada registro y la adjunta al elemento como una marca de tiempo de Beam adecuada, que es necesaria para la agrupación en períodos.

2. Por ejemplo:

| 'AddEventTimestamp' >> beam.Map(...)

Tarea 6: Agrupa elementos en períodos de un minuto

Pasemos a la segunda tarea de la secuencia de comandos, en la que transformaremos los elementos de los grupos según el desafío.

Minidesafío #TODO 2:

3. Esta transformación agrupa los elementos en períodos de tamaño fijo y sin superposición de 60 segundos (un minuto) según sus marcas de tiempo de eventos.

4. Por ejemplo:

| "WindowByMinute" >> beam.WindowInto(beam.window.FixedWindows(...))

5. Para obtener más información sobre otros tipos de renderización en ventanas, consulta la documentación de Apache Beam, en específico, la sección 8.2. Provided windowing functions.

Tarea 7: Registra eventos por período

Pasemos a la tercera tarea de la secuencia de comandos, en la que contaremos los eventos por período.

Minidesafío #TODO 3:

6. Este combinador cuenta la cantidad de elementos en cada período de un minuto. Usa .without_defaults() para asegurarte de que no se genere ningún resultado para los períodos vacíos.

7. Por ejemplo:

| "CountPerMinute" >> beam.CombineGlobally(CountCombineFn())...()

8. Para completar esta tarea, agrega una transformación que cuente todos los elementos de cada ventana. Recuerda consultar la solución si no puedes avanzar.

Tarea 8: Vuelve a convertir los elementos en filas y agrega una marca de tiempo

Pasemos a la última tarea de la secuencia de comandos, en la que volvemos a convertir los elementos en filas y agregamos una marca de tiempo.

Minidesafío #TODO 4:

9. La función GetTimestampFn (definida en pipeline_utils.py) toma el recuento de números enteros para cada período y lo convierte en formato de diccionario, agregando la hora de inicio del período como una cadena para que coincida con el esquema de BigQuery.

10. Para completar esta tarea, escribe una función ParDo que acepte elementos de tipo int y pase el parámetro adicional para acceder a la información de los períodos. Ten en cuenta que el campo de marca de tiempo en el esquema de la tabla de BigQuery es una CADENA, por lo que tendrás que convertir la marca de tiempo en una cadena.

11. Por ejemplo:

| "AddWindowTimestamp" >> beam.ParDo(...()) | 'WriteToBQ' >> beam.io.WriteToBigQuery( table_name, schema=table_schema, create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED, write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE )

Tarea 9: Ejecuta la canalización

Cuando completes la programación, ejecuta la canalización con el comando que aparece más abajo. Ten en cuenta que, mientras pruebas tu código, será mucho más rápido cambiar la variable de entorno RUNNER a DirectRunner, que ejecutará la canalización de manera local. Por ahora, ejecutaremos la canalización con Dataflow.

• Reemplaza los valores regionales y zonales según las especificaciones de tu lab.

Región

export PROJECT_ID=$(gcloud config get-value project) export REGION=ENTER_REGION_ID export BUCKET=gs://${PROJECT_ID} export PIPELINE_FOLDER=${BUCKET} export RUNNER=DataflowRunner export INPUT_PATH=${PIPELINE_FOLDER}/events.json export TABLE_NAME=${PROJECT_ID}:logs.minute_traffic python3 batch_minute_traffic_pipeline.py \ --project=${PROJECT_ID} \ --region=${REGION} \ --staging_location=${PIPELINE_FOLDER}/staging \ --temp_location=${PIPELINE_FOLDER}/temp \ --runner=${RUNNER} \ --input_path=${INPUT_PATH} \ --table_name=${TABLE_NAME} \ --setup_file=./setup.py

Tarea 10: Verifica los resultados

• Para completar esta tarea, espera unos minutos mientras se ejecuta la canalización y, luego, navega a BigQuery y consulta la tabla minute_traffic.

Haz clic en Revisar mi progreso para verificar el objetivo. Agrupar el tráfico del sitio por minuto y ejecutar la canalización

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.