TerramEarth — Vista técnica a los casos de estudio de la certificación Profesional Cloud Architect

Published in

Google Developer Experts

18 min readMar 23, 2019

El objetivo de este articulo no es evitar que tengas que estudiar por tu cuenta los casos de estudio, sino para ayudarte a ver los múltiples aspectos que se deben tener en cuenta al momento de analizar un proyecto de gran envergadura y de paso ver las implicancias técnicas detrás de cada uno de los componentes de la solución. Si sigues los pasos en tu cuenta gratuita de GCP te ayudará a estar mejor preparado para la certificación.

Encuentra el código en Github https://github.com/develasquez/casos-de-estudio/tree/master/TerramEarth

Caso de Estudio

Lo primero que debes hacer es analizar el caso de estudio de TerramEarth. Este fue recientemente revisado para la actualización del examen que se realizo en Noviembre del 2018.

Para resumir, TerramEarth cuanta con una gran flota de vehículos Agricola/Mineros los cuales generan TB de datos por día, el 20% de estos vehículos pueden enviar estas métricas mediante conexión inalámbrica, mientras que el resto es enviado cuando el vehículo entra en mantención.

La arquitectura de esta empresa esta dividida en dos flujos, el Batch y el Streaming, quedando algo similar a la siguiente imagen. Recuerda que esta es una solución tentativa ya que existen muchas forma de implementarla, te invito a ponerla a prueba y encontrar una mejor, te será de mucha ayuda para la Cert.

Hora te dejare un listado con cada uno de los componentes principales de la solución y un laboratorio (Codelabs o Qwicklab) para que lo conozcas más de cerca.

Cloud Storage
Functions (También serviría pero es muy caro para TerramEarth)
Dataflow Batch
BigQuery
Data Studio
BigQuery ML
IoT Core 1
IoT Core 2
Pub/Sub
Dataflow Streaming

No hagas trampa, deja de leer y termina los laboratorios XD.

Del papel a la Nube

Si ya hiciste los labs estas en condiciones de entrar en materia, vamos a hacer un análisis de cada uno de los pasos necesarios para llevar a TerramEarth a la Nube.

Lo primeros establecer el nombre de tu proyecto en GCP para facilitar las cosas:

TU_PROYECTO=xxxxxxxx

1) Pre Transferencia

Para el caso de los vehículos que se encuentran desconectados de la red, se espera un inmenso volumen de datos diarios, es por eso que es necesario comprimir los datos antes de subirlos a la nube.

Para ellos utilizaremos datos de ejemplo basados en el esquema de snon, puedes ver el archivo example.data.json a modo de ejemplo.

Para emular los datos generados por los vehículos, puedes ejecutar el script generateRandomMetrics.js, este generará un archivo llamado data.json, con 90000 registros de unos 120 campos cada uno, un total aprox de 312 MB.

#debes tener node.js instalado
node generateRandomMetrics.js > data.json

Recuerda que el punto importante en esta etapa es comprimir los datos para reducir los tiempos de transferencia, para ello utilizaremos gzip los que generará un archivo llamado data.json.gz que pesará unos 61.3 MB, una reducción superior al 80% del tamaño original. Se puede esperar los mismo en mayores volúmenes de datos, para el caso real de TerramEarth.

gzip data.json

Ok, ya tenemos los datos listos para subir a la nube, a jugar!!.

2) Transferencia

Transferencia Batch

Excelente ahora subamos esos datos, pero el método de transferencia no es un juego, y esto es muy importante de cara al examen. Ten en cuenta que para el caso de TE (TerramEarth) se van a acumular unos 891 TB por día y debemos tomar una importante decisión, ¿Qué mecanismo de transferencia utilizaremos? veamos qué nos ofrece Google.

Transfer Appliance Este método de transferencia consiste en que Google te envíe un Pendrive de unos 100 o 480 TB XD, fuera de broma, es un dispositivo rackeable, en el que puedes cargar tu data y enviarla de forma física y segura a Google Cloud. Esto ahorra mucho tiempo de carga, a un costo compuesto entre el servicio y el transporte desde el país de origen a Google.

Para el caso de TE esta solución no aplica ya que este sistema es para cargas One Time, pero TE necesita subir casi todos los días.

Storage Transfer Service Este método de transferencia permite importar datos desde sistemas online, los cuales pueden ser Amazon S3, Google Cloud Storage o un origen HTTP o HTTPS hasta un Google Cloud Storage dentro de tu proyecto.

Para nuestra solución este mecanismo tampoco nos sirve ya que los datos se encuentran en los servidores físicos de TE y no sería óptimo exponerlos por HTTP/S solo para poder transferirlos con este mecanismo.

gsutil Esta herramienta es muy versátil y poderosa, esta desarrollada en python y te da control absoluto de las acciones sobre Google Cloud Storage.

Lo que debes tener en cuenta es la velocidad de tu conexión a la red, el volumen de datos y el tiempo que dispones para subirla. Para ello TE debe utilizar el servicio de Cloud Interconnect, y elegir un tipo de conexión.

Dale un vistazo a las dos modalidades de interconnect

Dedicated Interconnect
Partner Interconnect

Imaginemos que TE se va por Dedicated interconnect con una velocidad de entre 10 Gbps y 80 Gbps (Máximo permitido). Ahora tengamos las siguientes consideraciones, TE genera 981 TB diarios de datos, si estos son comprimidos con gzip se reducirá teóricamente en un 80%, quedando un total de 196.2 TB comprimidos dependiendo del tipo de conectividad, podría demorar entre 60 horas y 6 horas en el mejor de los casos (con 80 Gbps)

Pero no basta con solo tener una buena velocidad, sino que hay estrategias para optimizar la transferencia, en este caso la más útil es la llamada parallel_composite_upload_threshold, esto cortará tus archivos en pequeños chuncks, para aprovechar el envío en paralelo, lo que reduce por mucho el tiempo de subida.

Para hacer la prueba, creemos un Bucket en nuestro proyecto, recuerda que el nombre debe ser único, reemplaza las XXXX por algo mágicamente único.

BUCKET_NAME=terramearth-batch-XXXX
gsutil mb gs://$BUCKET_NAME

Ahora debes dar un valor a parallel_composite_upload_threshold en MB, para nuestro ejemplo probemos con 15MB.

gsutil -o GSUtil:parallel_composite_upload_threshold=15M cp ./data.json gs://$BUCKET_NAME

Esto va a crear múltiples hilos, los cuales subirán nuestro archivo de forma paralela en pequeños chunks de 15MB, realmente hermoso verdad? XD.

Ahora veamos cómo sería este proceso para los vehículos que tiene conexión a internet.

Transferencia Streaming

Dentro de la flota de TE existe un 20% de los vehículos que cuentan con acceso a la red, lo que evita la acumulación de datos, y la necesidad de un proceso masivo. Muy por el contrario permite que estos datos se puedan procesar en streaming, cada vez que se van generando las muestras de sensores en los vehículos estos son enviados en tiempo real a la nube.

Para esto debemos comprender el concepto de IoT (Internet of Things), el cual busca estandarizar la forma en la que los dispositivos/ vehículos/ electrodomésticos se comunican y se gestionan a través de la red.

Dentro de los protocolos más utilizados para esto se encuentran el MQTT y el HTTP, y el componente que nos permite consumir estos en Google Cloud es Cloud IoT Core

Su funcionamiento en el caso de TE es bi-direccional, ya que permite recopilar los datos desde los vehículos, así como enviar nuevas configuraciones a estos.

Como se aprecia, estos datos en binario viajan haciendo uso de un tópico en Pub/Sub los que crearemos en unos instantes.

Para crear un registro de IoT core dentro de Google Cloud y poder hacer pruebas con este, puedes utilizar el ejemplo que se encuentra en la carpeta IoT de este repositorio.

Dado que la seguridad es primordial en la nube, los dispositivos que quiera comunicarse con Cloud IoT core deben hacer uso de tokens de JWT los que deben incluir una clave privada la cual es validada contra la llave pública almacenada en la configuración de IoT Core.

Para crear tus certificados auto firmados, puedes ejecutar el siguiente comando, aquí te dejo algo de documentación

#si es que estas en otro directorio
cd IoT/resources; openssl req -x509 -nodes -newkey rsa:2048 -keyout rsa_private.pem -days 1000000 -out rsa_cert.pem -subj "/CN=unused"

Lo primero que debes tener en cuenta es que para Cloud IoT Core solo tienes disponibles tres regiones, us-central1, europe-west1, and asia-east1.

Recuerda que para este caso los tópicos los debes crear antes que el registro de IoT Core. Estos tópicos serán los encargados de recibir como eventos cada uno de los mensajes que genere el dispositivo.

gcloud pubsub topics create te-tractor-topic;
gcloud pubsub topics create te-tractor-state-topic;

Cloud IoT Core permite la creación de registros para concentrar múltiples dispositivos con un objetivo o operativa en común, en este caso crearemos el registro para los tractores.

gcloud iot registries create te-tractor \
    --project=${TU_PROYECTO} \
    --region=us-central1 \
    --event-notification-config=topic=te-tractor-topic \
    --state-pubsub-topic=te-tractor-state-topic

Ahora debemos crear el dispositivo, es decir, un tractor en particular.

gcloud iot devices create te-tractor-device \
  --project=${TU_PROYECTO} \
  --region=us-central1 \
  --registry=te-tractor \
  --public-key path=rsa_cert.pem,type=rs256

Para emular los datos generados por el tractor he modificado el código de ejemplo en NodeJs de IoT Core en Github, éste toma el template de los 120 campos en un JSON y los envía por MQTT hacia IoT Core, que finale mente, los inyecta en el tópico que creamos en Pub/Sub

#vuelve al directorio TerramEarth/IoT
cd ..; 
#instalamos las dependenciasnpm install
 
# Emulamos en envío de 10 mensajes desde el tractor, puedes cambiar la cantidad pero creo que con 10 se entiende el concepto.node cloudiot_mqtt_example_nodejs.js mqttDeviceDemo    \
  --projectId=${TU_PROYECTO} \
  --cloudRegion=us-central1 \
  --registryId=te-tractor  \
  --deviceId=te-tractor-device  \
  --privateKeyFile=resources/rsa_private.pem \
  --numMessages=10 \
  --algorithm=RS256

Esto funciona de maravillas, aun que no tengas como verlo XD, si quisieras hacerlo, te recomiendo lo siguiente, Crea un flujo en DataFlow usando un template desde PubSub hacia Cloud Storage, esto creara un Flujo en streaming que tomara los eventos enviados y los dejara en un archivo dentro de un bucket. Dado que no es el funcionamiento final que esperamos, no documentare el proceso, pero funciona excelente y te animo a probarlo por tu cuenta, en especial considerando que a esta altura estamos ciegos respecto a los mensajes que están llegando al tópico.

Excelente ya logramos sacar los datos desde nuestros Tractores, tanto conectado como desconectados, pero para TE esto no es baratito, en realidad para ese volumen de datos es bastante caro.

Por ahora veremos el costo del proceso en streaming y a continuación veremos cómo abaratar los costos del proceso batch.

Hablemos de plata

Lo primero que tienes que tener presente es que si usas Cloud IoT Core con Cloud Pub/Sub, también se te facturará el consumo de recursos de Cloud Pub/Sub por separado.

Y considerando que TE genera 9TB de datos por día, podemos entender que generará un total de 279TB mensuales, si los agregamos a la calculadora de precios de Google Cloud, no antes de sumar el volumen de datos que se transmitirán por Pub/Sub que también son 279TB, lo que da un total de 153,841.30 USD, Wooow, mas de 150 mil dólares, solo por el proceso en Streaming, en la imagen a continuación puedes ver el detalle de cada uno de los componentes.

A mi parecer es muy caro, me hace pensar en que tal vez esos datos son descomprimidos, sin embargo una de las grandes ventajas de MQTT es que es binario y comprime el payload cerca del 85%, lo que baja un poco el total transferido, buscando info encontré un análisis de MQTT y su consumo y justo hay una estimación de 100 propiedades enviadas cada 1 hora, lo que da un aprox de 1.4MB mensual por dispositivo. Si lo multiplicamos por 4 millones de dispositivos (20% del total de la flota) da 5.6TB, si recordamos que esto es para 100 campos le agregamos el 20% lo que da 6.72TB mensuales a través de MQTT sobre Cloud IoT Core y llegando a Pub/Sub probablemente ya descomprimido. Así que hagamos otro cálculo.

Esto sí me hace más sentido, ya que de la otra forma IoT Core se vuelve inviable, creo, para cualquier compañía. Así que tengamos en cuanta en nuestras estimaciones que los datos se reducen al enviarlos por MQTT. Te dejo los dos pantallazos que me ayudaron a descubrir esto.

Y de este análisis genial en Device Wise

Ahora veamos cómo podemos optimizar los costos para el proceso Batch que es 4 veces más grande que el Streaming.

3) Almacenamiento de Archivos

Cada vez que tenemos que almacenar algo en la nube es muy importante que escojamos bien el tipo de almacenamiento que utilizaremos, este puede ser una Base de Datos en distintos tipos, un NFS y hasta un sistema de almacenamiento global como Cloud Storage.

Te dejo el link a la documentación oficial y un diagrama de flujo excelente que te ayudará a determinar que tipo de almacenamiento requiere tu solución, apréndetelo para la certificación te servirá mucho.

Para el caso de TE utilizaremos claramente Cloud Storage, pero como ya sabrás existen 4 clases de almacenamientos en este producto y una serie de buenas practicas que nos permitirán ahorrar unas moneditas XD.

Te dejo un link a las buenas prácticas que debes considerar al utilizar Cloud Storage.

El precio y características de cada una de las clases es el siguiente

Si te preocupan esos SLA, ten presente que los Storages de Googles están diseñados para cumplir un 99.999999999% (11 9’s) de durabilidad anual, Logrado esto gracias al almacenamiento redundante de objetos en múltiples dispositivos a lo largo de multiples zonas disponibles.

Estrategia de Storage TerramEarth

A mí parecer la mejor estrategia para este caso es un Storage Regional el cual tendrá un costo de $0.020 por Gb almacenado al mes, y una política de ciclo de vida que permita eliminar los archivos. Pero vamos por parte…

Tipo

Lo primero que tenemos que hacer es cambiar el tipo de Storage desde Multi-Regional a Regional, la docu aquí

gsutil rewrite -s regional -r gs://$BUCKET_NAME/**

Según la documentación debía ser así per me da error:

BadRequestException: 400 The combination of locationConstraint and storageClass you provided is not supported for your project

Así que la mejor solución a esta altura es crearlo de cero directamente regional.

# Para eliminar el bucketgsutil rm -r gs://$BUCKET_NAME# Para crearlo regional en gsutil mb -c regional -l us-central1 gs://$BUCKET_NAME/# Subimos nuevamente el Archivogsutil -o GSUtil:parallel_composite_upload_threshold=15M cp ./data.json gs://$BUCKET_NAME

ya tenemos el bucket en una región, la misma que IoT Core us-central1, lo que nos queda es crear una política para que elimine los archivos.

Politica

Creo que lo mejor es conservarlo 2 días, en caso que el proceso Batch no funcione a la primera. Para ello debemos crear un archivo JSON con el siguiente contenido, creo que se explica solo. Y como siempre aquí te dejo la documentación

{
	"lifecycle": {
	  "rule": [
	  {
	    "action": {"type": "Delete"},
	    "condition": {
	      "age": 2,
	      "isLive": true
	    }
	  }
	]
	}
}

Para aplicar la política debemos usar el siguiente comando:

gsutil lifecycle set lifecycle.json gs://$BUCKET_NAME

Esto nos permitirá controlar un poco el costo ya que los archivos durarán máximo 2 días. Igualmente esto nos mueve un tanto la aguja en los costos así que veamos cuánto sale tener estos datos en Cloud Storage.

Costos

Recuerda que TE genera un total de 196.2TB comprimidos diarios, y si estos se mantienen por dos días normalmente tendrás el doble de datos almacenados normalmente.

Si calculamos el costo de estos dato, 392.4TB permanentemente almacenados por mes en storage regional nos da:

La pequeña suma de 8 mil dólares XD, de todas formas podría haber sido más caro si no hubiéramos aplicado la compresión, el cambio de clase y la política de borrado automático.

4) Almacenamiento de Datos

Ante de mover los datos desde Pub/Sub y Cloud Storage, debemos pensar en el almacenamiento definitivo de los datos para su análisis, dentro de todos los mecanismos de almacenamiento que tenemos en GCP el más indicado para cumplir con el requerimiento de TE es BigQuery. pero ¿por que?, que preguntas te debes hacer para determinar esto.

Te doy un par de trucos:

Mira el diagrama de flujo que puse antes en la sección de Almacenamiento de Archivos.
La relación de los datos:
a) Si relacional, Cloud Sql, BigQuery, Spanner.
b) Si es No Relacional, Datastore, BigTable, Firestore.
Piensa en el volumen de datos:
a) Si es mayor a 10TB descarta Cloud SQL.
b) Petabytes? BigQuery, Spanner, BigTable.
El uso o consumo:
a) Si necesitas análisis, descarta Datastore, es poco queriable, no tiene funciones de agregación SUM, AVG, etc, no tiene OR en los Where y tampoco IN (<- esta es pregunta de cert) y los debes pedir uno a uno.
b) Para analítica? por excelencia es BigQuery.
c) Realtime? usa firestore.
d) TimeSeries? usa BitTable.
Con que se conecta:
a) Lo debes conectar a Data Studio, entonces piensa en Cloud Sql o BigQuery
b) Una App Mobile? Firestore por excelencia
Replicación:
a) Una Zona? Todas
b) Multi Zona? Cloud SQL
c) Multi Regional? Por excelencia Spanner por la altísima consistencia, pero es caro caro, BigQuery (EU, US), En Roadmap Cloud SQL, por ahora no, si es No SQL, firestore y Datastore.

Por lo tanto, si el volumen de datos de TE es tan grande, debe ser accedido desde ambas costas de US y su objetivo principal es el análisis. Entonces creo que la alternativa es Big Query.

Big Query: Esta es una poderosa Base de Batos muy similar a Apache Hive, la que permite consultar sobre Petabytes de datos en segundos. Se podría decir demasiado acerca de esta base de datos, pero no es el objetivo. Te invito a que veas sus características y limitaciones de cara a la certificación. Pero lo que tienes que saber es qué Google le esta poniendo todo su cariño a este motor y busca convertirla en un referente del mercado y eso nos conviene mucho. BigQuery hace que el trabajo sea muy fácil para nosotros y se integra a la perfección el resto de la plataforma GCP.

Manos a la Obra

Una de las grandes maravillas de BigQuery es que permite crear nuestros esquemas de base de datos de forma automática, tomando como base un archivo, él que puede ser CSV, JSON, Avro entre otros. E incluso lo hace si este esta comprimido. Una maravilla XD.

Para crear el esquema de forma automática a partir de el archivo JSON que dejamos en Google Cloud Storage basta con ejecutar el siguiente comando.

#Creamos el datasetbq --location=US mk --dataset --description "Dataset principal de TerramEarth" ${TU_PROYECTO}:terramearth#Creamos la tabla a partir del archivobq --location=US load --autodetect --source_format=NEWLINE_DELIMITED_JSON terramearth.tractordata gs://$BUCKET_NAME/data.json.gz

Las ventajas de esto es que no debes complicarte creando el esquema a mano, en especial cuando es tan complejo como es que queremos almacenar nosotros para TE.

Algo que tienes que tener en cuenta es que los datos deben estar en JSON, pero delimitados por un salto de línea, es decir, no es un Array con muchos objetos dentro separados por coma, sino un archivo que tiene un objeto JSON válido por cada línea.

Para ser un buen Arquitecto Cloud debes tener muchas consideraciones en especial con BigQuery, si piensas en la cantidad de datos almacenados, y las veces que se va a consumir esto puede costarle muy caro a TE. Así que veamos una serie de factores que nos ayudarán a optimizar esos costos. Te dejo un articulo hermoso aquí!!!, y las buenas prácticas oficiales acá.

Trata de no transformar datos con la query.
Usa aproximación en las funciones de agregación por ejemplo, en vez de COUNT(DISTINCT), usa APPROX_COUNT_DISTINCT().
Aplica los filtros antes de ordenar así el ordenamiento se hace sobre menos data.
Clusteriza tus Tablas.
Trata de buscar campos relevante para la clusterización, Fechas o grupos grandes.
Obliga a que se indique los valores del cluster en la query con require_partition_filter=true.
En la querys respeta el orden de los clusters.
Normaliza las tablas y usa correctamente los JOINS.
Usa GROUP BYs en base a los campos clusterizados.
Cuotas, puedes establecer límites en Kilobytes tanto a nivel de Proyectos, Datasets, Usuarios o Grupos en incluso por query.

Si somos descuidados con los precios podemos tener un desastre financiero, y ha pasado XD.. Fíjate cuanto saldría si consideramos los 900TB por día, por los 31 días del mes, tanto insertados, almacenados y queryados, por solo 1 mes (27.9 Petabytes)… Da más de 2 millos de dólares, lo que lo vuelve in viable.

Para que esta locura no ocurra sigue las buenas prácticas.

La solución al problema

Algo maravilloso que incluyo hace poco BigQuery es la Tarifa Plana, seguramente a TE le convenga mucho este enfoque. Tienes desde Flat Rate 40 con 2000 slots hasta Flat Rate 100 con 5000 slots ($100.000 USD / Mes).

Lo único que tendrás que pagar adicional es el almacenamiento, que en este caso para los 27 PB son como $600000 USD. Lo que nos hace pensar, en qué no se debe almacenar todos los datos, sino solo lo que sirvan para optimizar la compra de repuestos.

Otra cosa que recomiendo de cara a la certificación es que estudies los permisos necesarios para usar BigQuery tanto dentro de tu mismo proyecto como desde un proyecto distinto (<- esto me lo preguntaron en la certificación)

Bueno ya tenemos nuestra Tabla en la Base de Datos y solo nos queda mover los datos desde Pub/Sub y Cloud Storage a BigQuery.

5) Procesamiento

Venimos excelente con nuestra implementación profesional de TerramEarth, y no será menos en el procesamiento de datos, para ello haremos uso de Dataflow, tanto para el proceso Batch como para el basado en eventos con Pub/Sub

La mejor ventaja que nos da Cloud Dataflow es que tiene plantillas que abordan los escenarios más comunes de movimiento y transformación de datos.

Para conocerlos mejor entra a la documentación oficial de los Templates. Los que más nos sirven para este caso son:

Cloud Storage to BigQuery — Streaming: Este es un job en Dataflow que lee desde un origen en Cloud Storage, tomando un archivo que este en formato JSON, que incluso puede estar comprimido y lo transforma usando UDF para insertarlo en una tabla de BigQuery. La gran ventaja de este template que es en streaming o para entenderlo mejor, queda corriendo y en base a un patron identifica cuando se agrega un archivo al bucket y lo procesa inmediatamente.

Tomando en consideración la documentación de UDF, para poder implementar este Flujo primero se debe generar el esquema de destino en la Base de Datos, otra vez BigQuery nos hace el trabajo muy fácil. Para obtener este esquema basta con ejecutar:

bq show --format=prettyjson ${TU_PROYECTO}:terramearth.tractordata | jq '.schema.fields'

Tomamos el contenido que nos entrega este comando y lo guardamos en un archivo llamado schema.json y luego lo subimos a un Storage.

cd DataFlow;gsutil mb gs://${BUCKET_NAME}-dataflowgsutil cp schema.json gs://${BUCKET_NAME}-dataflow

Y Una función de conversión UDF que en este caso no hace mucho. Pero si quires reducir los datos que insertas en BQ este es el lugar indicado…

gsutil cp transform.js gs://${BUCKET_NAME}-dataflow/

Una vez que tienes los archivos arriba ejecutaremos el Job de Dataflow y lo dejaremos corriendo en espera de nuevos archivos.

JOB_NAME_GCS=gcs_text_to_bigquery-`date +"%Y%m%d-%H%M%S%z"`gcloud dataflow jobs run ${JOB_NAME_GCS} \
    --gcs-location gs://dataflow-templates/latest/Stream_GCS_Text_to_BigQuery \
    --parameters \
javascriptTextTransformFunctionName=transform,\
JSONPath=gs://${BUCKET_NAME}-dataflow/schema.json,\
javascriptTextTransformGcsPath=gs://${BUCKET_NAME}-dataflow/transform.js,\
inputFilePattern=gs://${BUCKET_NAME}/*.json.gz,\
outputTable=${TU_PROYECTO}:terramearth.tractordata,\
bigQueryLoadingTemporaryDirectory=gs://${BUCKET_NAME}-dataflow/temp

Asi se verá en Dataflow

Y para poder probar si funciona, debemos subir un archivo con nuevos datos, en este caso será el mismo pero con otro nombre, ya que lo que espera el flujo es un archivo que cumpla con el esquema JSON y con el patron de nombre *.json.gz.

gsutil cp ./data.json.gz gs://$BUCKET_NAME/data_2.json.gz

Si vamos a BigQuery y sacamos un Count deberíamos ver los 90000 registros iniciales y al finalizar el proceso deberíamos tener el doble.

bq query --nouse_legacy_sql 'select count(1) from `${TU_PROYECTO}.terramearth.tractordata`';

Cloud Pub/Sub to BigQuery — Streaming Este es un Job un poco más sencillo, va a leer el tópico de Pub/Sub en el que escribe Cloud IoT Core y lo va a insertar en una tabla de BigQuery, al escuchar un tópico de Pub/Sub este proceso queda activo esperando datos en Streaming.

Para poder implementarlo es muy sencillo. Ejecuta el siguiente comando:

JOB_NAME_PUB_SUB=pubsub-to-bigquery-`date +"%Y%m%d-%H%M%S%z"`
gcloud dataflow jobs run ${JOB_NAME_PUB_SUB} \
    --gcs-location gs://dataflow-templates/latest/PubSub_to_BigQuery \
    --parameters \
inputTopic=projects/${TU_PROYECTO}/topics/te-tractor-topic,\
outputTableSpec=${TU_PROYECTO}:terramearth.tractordata

Asi se ve en Dataflow:

Y cómo podemos probar que funciona???, Fácil, solo hay que ejecutar el proceso de envío de eventos a IoT Core que usamos al comienzo. (Dale unos minutos para el el Flujo levante los workers)

#vuelve al directorio TerramEarth/IoT
cd ../IoT; # Emulamos en envío de 10 mensajes desde el tractor, puedes cambiar la cantidad pero creo que con 10 se entiende el concepto.node cloudiot_mqtt_example_nodejs.js mqttDeviceDemo    \
  --projectId=${TU_PROYECTO} \
  --cloudRegion=us-central1 \
  --registryId=te-tractor  \
  --deviceId=te-tractor-device  \
  --privateKeyFile=resources/rsa_private.pem \
  --numMessages=10 \
  --algorithm=RS256

Podemos ver el resultado en Big Query, debería haber aumentado en 10 la cantidad de registros.

bq query --nouse_legacy_sql 'select count(1) from `${TU_PROYECTO}.terramearth.tractordata`';

Viste que fácil es usar Dataflow con los templates, ahora bien si quieres dar tus primeros pasos en Dataflow te recomiendo seguir esta Guía.

Hasta aquí ya tenemos los datos en BigQuery, sin embargo debemos visualizarlos con Data Studio y Analizarlos con BigQuery ML para predecir la demanda de repuestos.