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

Conocimientos básicos sobre Java

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.

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.

Activa Google Cloud Shell

Google Cloud Shell es una máquina virtual que cuenta con herramientas para desarrolladores. Ofrece un directorio principal persistente de 5 GB y se ejecuta en Google Cloud.

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

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

   

2. Haz clic en Continuar.

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

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

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

gcloud auth list

Resultado:

Credentialed accounts: - @.com (active)

Resultado de ejemplo:

Credentialed accounts: - google1623327_student@qwiklabs.net

• Puedes solicitar el ID del proyecto con este comando:

gcloud config list project

Resultado:

[core] project =

Resultado de ejemplo:

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

Configura tu IDE

Para los fines de este lab, usará principalmente un IDE web de Theia alojado en Google Compute Engine. El IDE tiene el repositorio de labs clonado previamente. Se ofrece compatibilidad con el servidor de lenguaje Java y una terminal para el acceso programático a las API de Google Cloud mediante la herramienta de línea de comandos de gcloud, similar a Cloud Shell.

Para acceder al IDE de Theia, copie y pegue en una pestaña nueva el vínculo que se muestra en Qwiklabs.

NOTA: Es posible que deba esperar entre 3 y 5 minutos para que se aprovisione por completo el entorno, incluso después de que aparezca la URL. Hasta ese momento, se mostrará un error en el navegador.

El repositorio del lab se clonó en su entorno. Cada lab se divide en una carpeta labs con un código que debe completar y una carpeta solution con un ejemplo viable que puede consultar como referencia si no sabe cómo continuar. Haga clic en el botón File Explorer para ver lo siguiente:

También puede crear varias terminales en este entorno, como lo haría con Cloud Shell:

Para verificarlo, ejecute gcloud auth list en la terminal con la que accedió como cuenta de servicio proporcionada, que tiene exactamente los mismos permisos que su cuenta de usuario del lab:

Si en algún momento su entorno deja de funcionar, intente restablecer la VM en la que se aloja el IDE desde la consola de GCE de la siguiente manera:

Parte 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 1: 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

• 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 3_Batch_Analytics/labs # Download dependencies mvn clean dependency:resolve export BASE_DIR=$(pwd)

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, copia este archivo automáticamente en tu bucket de Google Cloud Storage, en gs://my-project-id/events.json.

• 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 2: Suma las páginas vistas por usuario

1. Abre BatchUserTrafficPipeline.java en tu IDE, que puedes encontrar en 3_Batch_Analytics/labs/src/main/java/com/mypackage/pipeline.

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.

2. El siguiente paso en la canalización es agregar los eventos por cada user_id único y contar las páginas vistas de cada uno. Una manera sencilla de hacerlo en Filas o en objetos con un esquema de Beam es usar la transformación Group.byFieldNames() y, luego, realizar algunas agregaciones en el grupo resultante. Por ejemplo:

purchases.apply(Group.<MyObject>byFieldNames("userId", "address"));

Esto mostrará una PCollection de filas con dos campos, “key” y “values”. El campo “key” en sí es una Row con el esquema <userID:STRING, address:STRING>, que representa cada combinación única de userID y address. El campo “values” es del tipo ITERABLE[ROW[MyObject]], que contiene todos los objetos de ese grupo único.

FieldName	FieldType
key	ROW{userId:STRING, streetAddress:STRING}
values	ITERABLE[ROW[Purchase]]

3. Esto es muy útil cuando se pueden realizar cálculos agregados en esta agrupación y nombrar los campos resultantes de la siguiente manera:

purchases.apply(Group.byFieldNames("userId") .aggregateField("itemId", Count.combineFn(), "numPurchases") .aggregateField("costCents", Sum.ofLongs(), "totalSpendCents") .aggregateField("costCents", Max.ofLongs(), "largestPurchases"));

Esto muestra una Row de “clave” y “valor”, a diferencia del plural “valores”.

FieldName	FieldType
key	ROW{userId:STRING}
value	ROW{numPurchases: INT64, totalSpendCents: INT64, largestPurchases: INT64}

Las transformaciones Sum y Count son perfectas para este uso. Sum y Count son ejemplos de transformaciones Combine que pueden actuar sobre grupos de datos.

Nota: En este ejemplo, puedes realizar agregaciones en cualquiera de los campos de Count.combineFn() o, incluso, en el campo de comodín *, ya que esta transformación simplemente cuenta cuántos elementos hay en el grupo completo.

4. El siguiente paso de la canalización es agregar los eventos por user_id, sumar las páginas vistas y, además, calcular agregaciones adicionales en num_bytes, por ejemplo, el total de bytes del usuario.

Para completar esta tarea, agrega otra transformación a la canalización que agrupe los eventos por user_id y, luego, realiza las agregaciones relevantes. Ten en cuenta la entrada, las clases CombineFn que usarás y los nombres que asignarás a los campos de salida.

Tarea 3: Compacta el esquema

En este punto, la transformación nueva debería mostrar una PCollection con el esquema <Key,Value>, tal como se mencionó anteriormente. Si ejecutas tu canalización tal como está, se escribirá en BigQuery como dos RECORDS anidados, pese a que, básicamente, solo haya una fila de valores en cada uno.

1. Para evitar esto, agrega una transformación Select de la siguiente manera:

purchases.apply(Group.byFieldNames("userId") .aggregateField("itemId", Count.combineFn(), "numPurchases") .aggregateField("costCents", Sum.ofLongs(), "totalSpendCents") .aggregateField("costCents", Max.ofLongs(), "largestPurchases")) .apply(Select.fieldNames("key.userId", "value.numPurchases", "value.totalSpendCents", "value.largestPurchases")

Esta transformación conservará los nombres de campo relevantes en el nuevo esquema compactado y quitará “key” y “value”.

2. Para completar esta tarea, agrega una transformación Select para compactar el esquema de tu nueva fila.

Nota: Recuerda cambiar la sugerencia del objeto en BigQueryIO.<CommonLog>write() a <Row> si aún no las hecho.

Tarea 4: Ejecuta tu canalización

• Regresa a Cloud Shell 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.

export PROJECT_ID=$(gcloud config get-value project) export REGION='us-central1' export BUCKET=gs://${PROJECT_ID} export PIPELINE_FOLDER=${BUCKET} export MAIN_CLASS_NAME=com.mypackage.pipeline.BatchUserTrafficPipeline export RUNNER=DataflowRunner export INPUT_PATH=${PIPELINE_FOLDER}/events.json export TABLE_NAME=${PROJECT_ID}:logs.user_traffic cd $BASE_DIR mvn compile exec:java \ -Dexec.mainClass=${MAIN_CLASS_NAME} \ -Dexec.cleanupDaemonThreads=false \ -Dexec.args=" \ --project=${PROJECT_ID} \ --region=${REGION} \ --stagingLocation=${PIPELINE_FOLDER}/staging \ --tempLocation=${PIPELINE_FOLDER}/temp \ --runner=${RUNNER} \ --inputPath=${INPUT_PATH} \ --tableName=${TABLE_NAME}"

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

Tarea 5: Verifica los resultados en BigQuery

1. Para completar esta tarea, espera unos minutos hasta que se complete la canalización. Luego, navega a BigQuery y consulta la tabla user_traffic.

2. Si te interesa, puedes marcar como comentario el paso de transformación Select y volver a ejecutar la canalización para ver el esquema resultante de BigQuery.

Parte 2: Agrupa el tráfico del sitio por minuto

En esta parte del lab, crearás una nueva canalización llamada BatchMinuteTraffic. BatchMinuteTraffic amplía los principios básicos del análisis por lotes que se usan en BatchUserTraffic y, en lugar de agrupar el tráfico por usuarios en todo el lote, lo agrupa cuando ocurren los eventos.

• En el IDE, abre el archivo BatchMinuteTrafficPipeline.java dentro de 3_Batch_Analytics/labs/src/main/java/com/mypackage/pipeline.

Tarea 1: Agrega marcas de tiempo a cada elemento

Una fuente no delimitada proporciona una marca de tiempo para cada elemento. Según la fuente no delimitada, es posible que debas configurar cómo se extrae la marca de tiempo del flujo de datos sin procesar.

Sin embargo, las fuentes delimitadas (p. ej., un archivo de TextIO, como el que se usa en esta canalización) no proporcionan marcas de tiempo.

1. Puedes analizar el campo de marca de tiempo desde cada registro y usar la transformación WithTimestamps para adjuntar las marcas de tiempo a cada elemento en tu PCollection.

PCollection<MyClass> unstamped = ...; PCollection<MyClass> stamped = unstampedLogs.apply(WithTimestamps.of((MyClass m) -> org.joda.time.Instant.parse(m.getTimestamp())));

2. Para completar esta tarea, agrega una transformación a la canalización que agregue marcas de tiempo a cada elemento de la canalización.

Tarea 2: Agrupa elementos en ventanas de un minuto

La renderización en ventanas subdivide una PCollection de acuerdo con las marcas de tiempo de sus elementos individuales. Las transformaciones que agregan varios elementos, como GroupByKey y Combine, funcionan de forma implícita por ventana, es decir, procesan cada PCollection como una sucesión de ventanas múltiples y finitas, aunque toda la colección en sí puede ser de un tamaño ilimitado.

Puedes definir diferentes tipos de ventanas para dividir los elementos de tu PCollection. Beam proporciona varias funciones analíticas, incluidas las siguientes:

• Ventanas de tiempo fijo
• Ventanas de tiempo variable
• Ventanas por sesión
• Ventana global única
• Ventanas basadas en el calendario (no compatibles con el SDK de Beam para Python)

En este lab, utilizarás ventanas de tiempo fijo. Una ventana de tiempo fijo representa un intervalo de tiempo que no se superpone y de duración coherente en el flujo de datos. Considera ventanas con una duración de cinco minutos: todos los elementos de tu PCollection no delimitada con valores de marca de tiempo desde 0:00:00 hasta (sin incluir) 0:05:00 pertenecen a la primera ventana. Los elementos con valores de marca de tiempo desde 0:05:00 hasta (sin incluir) 0:10:00 pertenecen a la segunda ventana, y así sucesivamente.

1. Implementa un período de tiempo fijo con una duración de un segundo como se indica a continuación:

PCollection<String> pColl= ...; PCollection<String> windowedPCollection = pColl.apply( Window.<String>into(FixedWindows.of(Duration.standardSeconds(60))));

2. Para completar esta tarea, agrega una transformación a tu canalización que agrupe los elementos en ventanas de un minuto.

Para obtener más información sobre otros tipos de sistemas de ventanas, lee la sección sobre las funciones analíticas en la documentación de Apache Beam.

Tarea 3: Calcula eventos por ventana

A continuación, la canalización debe calcular la cantidad de eventos que ocurrieron en cada ventana. En la canalización de BatchUserTraffic, se usaba una transformación Sum para sumar por clave. Sin embargo, a diferencia de esa canalización, en este caso los elementos tienen ventanas y el cálculo deseado debe respetar los límites de esas ventanas.

A pesar de esta nueva restricción, la transformación Combine aún es apropiada. Esto se debe a que este tipo de transformaciones respetan automáticamente los límites de las ventanas.

1. Consulta la documentación sobre Count para aprender a agregar una transformación nueva que cuente la cantidad de elementos por ventana.

A partir de Beam 2.22, la mejor opción para contar elementos de filas en el sistema de ventanas es usar Combine.globally(Count.<T>combineFn()).withoutDefaults() (es decir, no usar todo SQL, que analizaremos más adelante en el siguiente lab). Esta transformación dará como resultado el tipo PCollection<Long> que, como verás, ya no usa esquemas de Beam.

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

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

Para escribir en BigQuery, cada elemento debe volver a convertirse en un objeto Row con un campo llamado “pageviews” y un campo adicional llamado “minute”. La idea es usar el límite de cada ventana como un campo y la cantidad combinada de páginas vistas como el otro.

Hasta ahora, los elementos siempre se han ajustado a un esquema de Beam cuando pasan de ser una cadena JSON a ser un objeto CommonLog y, a veces, cuando se revierten a un objeto Row. Se dedujo el esquema original del POJO de CommonLog a través de la anotación @DefaultSchema(JavaFieldSchema.class) y, luego, se especificaron los campos agregados o borrados en las transformaciones de las canalizaciones. Sin embargo, en este punto de la canalización, según el resultado de la transformación Count, cada elemento es del tipo Long. Por lo tanto, se debe crear un objeto Row nuevo desde cero.

1. Los esquemas se pueden crear y registrar de forma manual como se describe a continuación. Este código se agregaría fuera del método main(), de manera similar a la definición del objeto CommonLog:

// Define the schema for the records. Schema appSchema = Schema .builder() .addInt32Field("appId") .addStringField("description") .addDateTimeField("rowtime") .build();

2. Los objetos Row posteriores de este esquema se pueden crear en una PTransform en función de entradas como una Long. Este es un ejemplo:

Row row = Row .withSchema(appSchema) .addValues(1, "Some cool app", new Date()) .build();

3. Por lo general, Beam requerirá una indicación del esquema nuevo en la PTransform si la transformación crea una fila nueva en lugar de mutar una anterior:

.apply().setRowSchema(appSchema)

Obtén más información sobre la creación y la inferencia de esquemas en la Explicación de Apache Beam SQL.

Otro problema en este punto es que la transformación Count solo proporciona elementos de tipo Long que ya no llevan ningún tipo de información de marca de tiempo.

Sin embargo, esta información sí está presente, aunque no de forma tan clara. Los ejecutores de Apache Beam saben de forma predeterminada cómo proporcionar el valor para varios parámetros adicionales, incluidas las marcas de tiempo de eventos, las ventanas y las opciones de canalización. Consulta la documentación de parámetros de DoFn de Apache para obtener una lista completa.

4. Para completar esta tarea, escribe una función ParDo que acepte elementos de tipo Long y emita elementos de tipo Row con el tipo de esquema pageViewsSchema, que se te proporcionó anteriormente y que tiene un parámetro de entrada adicional de tipo IntervalWindow. Utiliza este parámetro adicional para crear una instancia de Instant. Luego, usa esta instancia para derivar una representación de cadena para el campo de minutos":"

@ProcessElement public void processElement(@Element T l, OutputReceiver<T> r, IntervalWindow window) { Instant i = Instant.ofEpochMilli(window.start().getMillis()); ... r.output(...); }

Tarea 5: 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.

export PROJECT_ID=$(gcloud config get-value project) export REGION='us-central1' export BUCKET=gs://${PROJECT_ID} export PIPELINE_FOLDER=${BUCKET} export MAIN_CLASS_NAME=com.mypackage.pipeline.BatchMinuteTrafficPipeline export RUNNER=DataflowRunner export INPUT_PATH=${PIPELINE_FOLDER}/events.json export TABLE_NAME=${PROJECT_ID}:logs.minute_traffic cd $BASE_DIR mvn compile exec:java \ -Dexec.mainClass=${MAIN_CLASS_NAME} \ -Dexec.cleanupDaemonThreads=false \ -Dexec.args=" \ --project=${PROJECT_ID} \ --region=${REGION} \ --stagingLocation=${PIPELINE_FOLDER}/staging \ --tempLocation=${PIPELINE_FOLDER}/temp \ --runner=${RUNNER} \ --inputPath=${INPUT_PATH} \ --tableName=${TABLE_NAME}"

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

Tarea 6: Verifica los resultados

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

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.