Data Engineering Zoomcamp – Semana 3

La tercera semana del bootcamp para Data Engineers se ha centrado en Data warehouse y en concreto, el servidio de Google Cloud Platform dedicado a tal fin: BigQuery. Hemos abordado conceptos como particionamiento y clustering para optimizar el consumo de datos.

• Repo Bootcamp: https://github.com/DataTalksClub/data-engineering-zoomcamp/tree/main/week_3_data_warehouse
• Repo personal con ejercicios: https://github.com/hegdehog/data-engineering-zoomcamp-2023

Última actualización 06/02/2023

Prerequisitos

Esta semana no hemos dado grandes pasos en la suite tecnológica. Para seguir el temario y realizar los ejercicios continuaremos con la cuenta de Google Cloud Platform y la instancia de BigQuery que creamos a inicio del bootcamp.

Google BigQuery

Dentro de la suite de servicios cloud que ofrece la GCP (Google Cloud Platform) encontramos BigQuery como solución de almacenamiento y análisis de datos. Como todo servicio cloud, destaca por la escalabilidad, alta disponibilidad y pago por uso. Entre sus características, destacan:

1. Serverless: al ser un PaaS el usuario no debe preocuparse por el mantenimiento software/hardware, sólo de su configración y explotación.
2. Escalabilidad: BigQuery es altamente escalable, lo que significa que puede manejar un aumento en el volumen de datos sin interrupciones.
3. Almacenamiento en la nube: BigQuery permite almacenar y procesar grandes volúmenes de datos en la nube de Google, lo que significa que los usuarios no tienen que preocuparse por la infraestructura de hardware.
4. Análisis en tiempo real: BigQuery permite realizar consultas en tiempo real sobre grandes cantidades de datos.
5. Integración con herramientas de Google: BigQuery se integra con otras herramientas de Google, como Google Data Studio, Google Cloud Storage y Google Cloud Dataproc.
6. Consultas SQL: BigQuery permite realizar consultas SQL en tiempo real sobre grandes cantidades de datos. El SQL es compatible con la semántica de transacción de la base de datos (ACID)
7. Pago por uso (pay-as-you-go): BigQuery es un servicio de pago por uso, lo que significa que los usuarios solo pagan por el almacenamiento y el análisis de datos que utilizan.
8. Seguridad y cumplimiento normativo: BigQuery ofrece una amplia gama de características de seguridad y cumplimiento normativo para ayudar a proteger los datos de los usuarios.
9. Compatibilidad con diferentes formatos de datos: BigQuery es compatible con diferentes formatos de datos, como CSV, JSON y Avro, y permite a los usuarios importar y exportar datos de y hacia otras plataformas.

Precios de BigQuery

La facturación de BigQuery consta de dos partes: el precio del análisis y el del almacenamiento. Más información en el pricing de GCP BigQuery.

• El precio de análisis: coste de procesar consultas (SQL, funciones definidas por el usuario UDF, scripts DML y DDL). Sin entrar en mucho detalle, hay dos modelos de precios disponibles para el análisis: bajo demanda (5$ por TB) o tarifa plana (el precio varía en función del tipo ranura, mensual o anual)
• El precio de almacenamiento coste de almacenar físicamente los datos. El almacenamiento es más barato y encontramos dos tipos (similar a los tiers del los Storage Account de Azure):
  • Activo: 0,020$/GB (los primeros 10 GB/MES son gratis). Automáticamente para las tablas o particiones con cambios en los últimos 90 días.
  • Largo plazo: 50% de descuento: 0,010$/GB para aquellas tablas y particiones que no se hayan tocado en los últimos 90 días.

External tables

Al igual que en otras soluciones cloud como en Azure Synapse, desde Google BigQuery podemos crear tablas externas (external tables) para consultar conjuntos de datos como si fueran una tabla SQL que no estén almacenados en BigQuery (archivos en un Cloud Storage, otra base de datos fuera de GCP, etc). Este tipo de objeto almacena en BQ únicamente el esquema, que infiere del fichero a la hora de crear el objeto, pero no los datos. Las tablas externas tienen las mismas características que una tabla estándar en BigQuery, con sus propiedades, administración de acceso, metadatos, etc. La única diferencia es que son una cáscara, los datos están en otra ubicación.

Por ejemplo, en lugar de ingestar un CSV en una tabla de la BBDD BigQuery, vamos a crear una external table para acceder directamente a los datos sin persistir:

CREATE OR REPLACE EXTERNAL TABLE 'my_database.yellow_tripdata_external_table'
OPTIONS (
  format = 'CSV',
  uris = ['gs://data/trip data/yellow_tripdata_2020-01.csv']
);

Optimización de Google BigQuery

A diferencia de una base de datos relacional, BigQuery no soporta índices para agilizar las consultas SQL. En su lugar, ofrece dos alternativas: particionado (partitioning) y agrupación (clustering). Estas opciones no son recomendables cuando nuestro volumen de datos es < 1GB.

Particionado (Partitioning)

Una tabla particionada es una tabla dividida en segmentos aka particiones basados en los valores de una columna. Dividir una tabla en porciones agiliza enormemente las consultas ya que los datos que necesita recuperar son mucho más reducidos que si tuviera que leer toda la tabla. BigQuery ofrece tres tipos de particiones:

• Partición por rangos de números enteros: se crean particiones en base al valor numérico de una columna de tipo INTEGER. Por ejemplo, por el código de país.
• Partición de columnas por unidad de tiempo: la partición más habitual, se particiona la tabla por una columna de tipo DATE, TIMESTAMP o DATETIME.
• Partición por tiempo de ingesta: BigQuery asigna de forma automática filas a las particiones en función del momento en que BigQuery transfiere los datos. Puedes elegir el nivel de detalle por hora, día, mes o año para las particiones. Tiene un límite de 4k particiones. Se añade la columna _PARTITIONTIME y en cada tupla se asigna el valor del momento en el que se ha almacenado el dato.

En el siguiente ejemplo creamos una nueva tabla a partir de una consulta SELECT y añadimos el particionado por la columna tpep_pickup_datetime

CREATE OR REPLACE TABLE my_database.yellow_tripdata_partitoned
PARTITION BY
  DATE(tpep_pickup_datetime) AS
SELECT * FROM my_database.yellow_tripdata;

Cuando lanzas una query en BigQuery sobre una tabla particionada no es necesario que lo especifiques, internamente lo identifica y realiza la consulta sobre la partición que cumpla el criterio del predicado (o filtro). Diferencia importante respecto a SQL Server u Oracle, donde sí se debe especificar la partición en la sentencia SELECT.

SELECT DISTINCT(VendorID)
FROM my_database.yellow_tripdata_non_partitoned
WHERE DATE(tpep_pickup_datetime) BETWEEN '2021-01-01' AND '2021-02-28';

Agrupación (Clustering)

El clustering reordena los datos de la tabla en base a una o más columnas (hasta 4). Características de la agrupación de columnas en BigQuery:

• El orden de las columnas agrupadas es relevante para determinar la prioridad de las columnas
• Mejora el rendimiento en consultas que utilicen predicados o funciones de agregación
• Funciona mucho mejor con columnas con una alta cardinalidad (email, categorías, nombres)
• Se pueden hacer agrupaciones en columnas de tipo:
  • DATE
  • BOOL
  • GEOGRAPHY
  • INT64
  • NUMERIC
  • BIGNUMERIC
  • STRING
  • TIMESTAMP
  • DATETIME
• Límite de como máximo 4 columnas agrupadas (clusters) por tabla.

Podemos crear clusters al mismo tiempo que las particiones. Tomando como ejemplo la query anterior, vamos a añadir una columna grupada o cluster por el campo VendorID:

CREATE OR REPLACE TABLE my_database.yellow_tripdata_partitoned_clustered
PARTITION BY DATE(tpep_pickup_datetime)
CLUSTER BY VendorIDKey AS
SELECT * FROM my_database.external_yellow_tripdata;

BigQuery Partitioning vs Clustering

Podemos utilizar ambas opciones de optimización juntas y aprovechar las sinergias que generan, o en función de la necesidad elegir una estrategia u otra. Normalmente usaremos particionado cuando nuestras consultas filtran o agregan datos sobre un único campo con baja cardinalidad (año, mes, ciudad, departamento…), mientras que si es por varios podemos particionar por el que tenga menor cardinalidad (fecha, por ejemplo) y aplicar clustering por el resto (hasta un máximo de 4 columnas).

Partitioning	Clustering
El coste de la query es conocido. BigQuery puede estimar la cantidad de datos que va a recuperar antes de ejecutar la query.	Se desconoce el coste de la query ya que no puede estimar la cantidad de datos.
Granularidad baja. Sólo se puede aplicar partición por una columna.	Granularidad alta. Se pueden utilizar varias columnas para reordenar la tabla (hasta un máximo de 4)
Enfocada para queries que filtran o agregan datos por una sóla columna.	Enfocada para queries que filtran o agregan por varias columnas.
Límite de 4K particiones de una columna, lo que implica que sólo se puede utilizar con campos con baja cardinalidad (o hasta 4K).	No hay límite de clusters, por lo que soporta columnas con una alta cardinalidad.

El propio motor de BigQuery es el encargado de actualizar automáticamente los clusters cuando se insertan nuevos datos en la tabla de forma transparente para el usuario. En el caso de tablas particionadas, los clusters se actualizan de forma independiente en cada partición.

Buenas prácticas SQL para optimizar consultas

La mayoría no sólo aplican a Google BigQuery, se trata de recomendaciones para queries ejecutadas sobre cualquier motor de BBDD:

• Evitar usar SELECT *, lo ideal es recuperar sólo las columnas que necesitemos o vayamos a utilizar.
• Evalúa el coste de ejecutar la query antes de lanzarla. Esto es especialmente útil en entornos cloud donde la facturación seleccionada sea por ejecución (se paga por cada ejecución), que normalmente es más caro que si se selecciona una capacidad o paquete.
• Aplicar la optimización por partitioning y/o clustering
• En los casos de real-time debemos prestar atención y ser cuidadosos con los INSERT de datos (insertAll)
• Crear vistas materializadas como pasos intermedios cuando la query debe manejar un gran volumen de datos. Debemos tener en cuenta que BigQuery también cachea los resultados de las columnas.
• Aplicar filtros por columnas de partición o agrupación (clusters)
• Desnormalizar los datos bajando las formas normales a lo mínimo, en otras palabras, destruir la integridad referencial manteniendo todos los datos en una única tabla para evitar joins entre varias. Se recomienda usar campos anidados o repetidos (nested fields) con STRUCT o ARRAY. Aunque tiene ciertas desventajas (mayor almacenamiento por repetir datos y se pierde la integridad de los datos), es la forma más óptima de explotar grandes volúmenes de datos.
• Intenta utilizar funciones de agregación aproximada HyperLogLog++ o HLL++. Necesita menos memoria que las funciones de agregación exacta, como COUNT(DISTINCT), pero generan incertidumbre estadística. Son muy útiles para grandes volúmenes de datos en los que el uso de la memoria lineal es poco práctico teniendo en cuenta que el dato que nos devuelve es una aproximación estadística, no el valor exacto.
• Evitar usar funciones propias SQL o JavaScript UDF
• Cuando cruces varias tablas organiza el JOIN ubicando la más voluminosa primero. Será la que BigQuery use en primer lugar para distribuirla por los nodos y las siguientes tablas las irá repartiendo por cada uno. Además, intenta reducir el tamaño de las subqueries o vistas materializadas antes de hacer los cruces.

Arquitectura Google BigQuery

La arquitectura de BigQuery desvincula el almacenamiento de la computación (motor de análisis), lo que permite escalar cada recurso de forma independiente. Esta flexibilidad favorece un control de costes mucho más granularizado ¿Qué piezas encontramos dentro de BigQuery? Dremel, Colossus, Jupiter y Borg:

Borg: orquestador de contenedores

Orquestador de contenedores propio de Google que se encarga de proveer del hardware necesario para que operen los slots y mixers del motor Dremel.

Jupyter: red

Como la estructura de BigQuery es desacoplada (separa físicamente el almacenamiento del motor de cómputo) necesita un artefacto que conecte ambas entidades: Jupyter. Ofrece suficiente ancho de banda como para permitir la comunicación entre 100K máquinas a una velocidad ultra rápida de 10Gbs/s.

Dremel: motor de ejecución

Se trata del motor de consultas de alta velocidad de BigQuery que usa Google en su propio buscador. Orquesta las consultas segmentándolas en pequeñas porciones que se distribuyen por nodos y cuando finalizan se agrupan para devolver el resultado; la definición de procesamiento distribuido. Dremel convierte una consulta SQL en un árbol de ejecución donde encontramos slots y mixers, todos ejecutados sobre Borg (ver más adelante). El propio motor asigna de forma dinámica slots a las consultas que van entrando:

• Slots: serían las hojas del árbol y se encargan de la parte más pesada: leer datos en Colossus y realizar operaciones de cómputo.
• Mixers: las ramas. Se encargan de las operaciones de agregación

Colossus: almacenamiento distribuido

Sistema de almacenamiento distribuido de última generación de Google. Gestiona las réplicas, recuperación (cuando los discos fallan) y administración distribuida (lo que mitiga el impacto en caso de caída). Colossus utiliza el formato columnar y de compresión ColumnIO capaz de manejar fácilmente petabytes de datos.

Almacenamiento

Column-oriented vs record-oriented

Tradicionalmente el almacenamiento de datos tabular está orientado a filas (record-oriented o row-oriented). De esta forma, los datos se almacenan y consultan fila por fila y después se accede a cada una de las columnas. BigQuery en su lugar, utiliza almacenamiento orientado a columnas (column-oriented). Este tipo de almacenamiento es ideal para grandes volúmenes de datos ya que permite reducir la consulta sólo a las columnas deseadas, sin necesidad de recuperar todos los datos de una fila para luego quedarnos con lo que nos interesa.

El motor de ejecución Dremel guarda los datos en Colossus en formato columnar, es decir, divide un registro o fila por los valores de sus columnas y almacena cada valor en un volumen distinto. En las base de datos tradiciones normalmente se almacena el registro completo en el mismo volumen. Cuando se lanza una consulta sobre BQ, Dremel crea un árbol de ejecución distribuyendo la query en porciones a través de los mixers que a su vez se distribuyen en paquetes más pequeños sobre la malla de slots que acceden a la capa de almacenamiento Colossus (la siguiente sección incluye un diagrama al final donde se visualiza esta explicación).

Campos anidados y repetidos (repeated & nested fields)

Una de las buenas prácticas para modelar una base de datos analítica es desnormalizar con el fin de reducir los cruces entre tablas (hechos + dimensiones). Se almacenan todos los datos en una única tabla, de forma que la información (como la ciudad) se repite y almacena en la misma tabla, no es necesario cruzar con una tabla de dimensión para recuperar los valores. BigQuery soporta el uso de estructuras repetidas o anidadas mediante las operadores STRUCT y ARRAY. Esto nos permite representar la información de forma natural además de agilizar las consultas. Podemos combinar ambas estrategias para el almacenamiento de datos

El motor Dremel debe su velocidad, entre otras cosas, al almacenamiento columnar de los campos anidados (nested fields). Todos los valores de un campo anidado (por ejemplo A.B.C) se almacenan de forma contigua, por lo que permite que se recupere sólamente A.B.C sin necesidad de leer A.E, A.B.D, etc.

En la siguiente query vamos a crear la tabla my_database.tabla_array_struct con cuatro elementos: dos columnas estándar de tipo string y date, y dos campos anidados: cliente de tipo ARRAY y producto que sería un ARRAY o colección de STRUCT.

CREATE TABLE `my_dataset.tabla_array_struct`
(
  ID_PEDIDO int64,
  FECHA date,
  CLIENTE  ARRAY<string>,  
  PRODUCTOS ARRAY<STRUCT<SKU int64, CANTIDAD int64, PRECIO numeric>>,
);

Poblamos la tabla de la siguiente forma:

INSERT INTO `my_dataset.tabla_array_struct` (ID_PEDIDO, FECHA, CLIENTE, PRODUCTOS)
VALUES (
    1,
    CAST('2021-01-01' AS date), 
    ['231', 'Marcos Torregrosa', 'Madrid, Spain'],
    [
      STRUCT(6633 as SKU, 2 as CANTIDAD, cast(3.99 as numeric) as PRECIO),
      STRUCT(1667, 1, 1.99),
      STRUCT(8871, 1, 16.99)
    ] 
    );

Si observamos el esquema vemos que BQ trata estas columnas como RECORD y modo REPEATED.

Echando un ojo a los datos:

Con OFFSET u ORDINAL podemos acceder directamente por el índice a cualquier elemento de una columna de tipo ARRAY:

SELECT cliente, cliente[OFFSET(1)] as offset_1, cliente[ORDINAL(1)] as ordinal_1
FROM my_dataset.tabla_array_struct

Con el operador UNNEST podemos aplanar un ARRAY mostrar una tabla con una fila para cada elemento en ese array. 

select * from my_dataset.tabla_array_struct, UNNEST(cliente) as clientes
where cliente[OFFSET(1)] like "Marcos%"

BigQuery y Machine Learning

Desde BigQuery podemos ejecutar modelos de machine learning de forma sencilla y ágil usando simplemente SQL estándar sin necesidad de plataformas específicas, movimiento de datos o conocimientos de programación (python, scala, etc). Los algoritmos de ML disponibles nativamente dentro de BigQuery pueden consultarse en la documentación oficial.

Por ejemplo, para crear un modelo de regresión lineal con el objetivo de predecir la propina (tip_amount) de una carrera en taxi usamos la sentencia CREATE MODEL (recomendable ver la documentación con todas las opciones):

CREATE OR REPLACE MODEL `my_database.nytaxi.tip_model`
OPTIONS (
  model_type='linear_reg',
  input_label_cols=['tip_amount'],
  DATA_SPLIT_METHOD='AUTO_SPLIT'
) AS
SELECT
  *
FROM
  `my_database.nytaxi.yellow_tripdata`
WHERE
  tip_amount IS NOT NULL;

• CREATE OR REPLACE MODEL es la sentencia para crear nuestro modelo
• Dentro de las OPTIONS() vamos a indicar los parámetros y configuración del modelo:
  • MODELE_TYPE='linear_reg' en nuestro ejemplo vamos a crear un modelo de regresión lineal. Podríamos usar cualquiera de los disponibles en BQ (como KMEANS para crear clústers de datos o RANDOM_FOREST_CLASSIFIER para crear un modelo de clasificación)
  • INPUT_LABEL_COLS=['tip_amount'] Array de columnas separadas por comas que vamos a utilizar para entrenar y usar el modelo.
  • DATA_SPLIT_METHOD='AUTO_SPLIT' especificamos que queremos dividir automáticamente el conjunto de datos en dos partes, una de entrenamiento y otra de comprobación (training/test).
• El SELECT especifica la fuente de datos, así como el predicado si lo hubiera (filtro).

El modelo que hemos creado se almacena dentro del conjunto de datos en BigQuery como un nuevo objeto:

BQ nos ofrece una serie de sentencias para analizar y explotar el modelo. Más información en la documentación oficial.

• ML.FEATURE_INFO: muestra estadísticas de cada columna del conjunto de datos (valores mínimo y máximo, medias, etc). Similar a ejecutar el comando describe() de en Pandas (python).
• ML.EVALUATE: muestra las métricas de un modelo, ideal para comprobar con un nuevo conjunto de datos cómo respondería el modelo. las métricas que ofrece son las mismas que podemos consultar viendo el detalle del model creado desde el GUI de GCP.
• ML.PREDICT: nos permite ejecutar el modelo sobre un conjunto de datos y generar las predicciones para las que se haya configurado.
• ML.EXPLAIN_PREDICT: añade información a la sentencia anterior sobre cuáles de las columnas o features son las que más colaboran para el cálculo de la predicción.