Mejorando tu Datalake. Apache Iceberg al rescate.

Contexto y necesidades

Tenemos bastante asimilado el concepto de Datalake a día de hoy. Desde su aparición ha ido integrando diferentes tecnologías que han ido aumentando la capacidad de procesamiento y analítica. Siempre con un objetivo claro: simplicidad, flexibilidad y bajo coste.

Combinado estos tres factores con la aparición de tecnologías y formatos emergentes como Iceberg, hablamos de la evolución de Datalake hacia lo que se conoce como “open data management architecture” o Lakehouse. La arquitectura Lakehouse se está convirtiendo rápidamente en el nuevo estándar para plataformas de datos, analítica e IA, e intenta resolver las problemáticas que se encuentran en las plataformas de datos tradicionales. Las prioridades de las empresas pasan por liberar todo el potencial de los datos, tomar decisiones de manera ágil a partir de datos actualizados más frecuentemente, romper los silos y gobernar sus datos de forma eficiente.

En el post de hoy os contamos cómo ha sido nuestra experiencia práctica con un Lakehouse en Google Cloud, integrando Apache Iceberg como formato de alto rendimiento para tablas analíticas, Dataproc con Spark para DML y BigQuery.

Introducción a Apache Iceberg y capacidades

Apache Iceberg es un formato de alto rendimiento para tablas analíticas de gran tamaño. ¿En qué se diferencia de formatos de ficheros columnares como Apacher Parquet u ORC? Veámoslo.

Iceberg nace ante la necesidad de darte cierta capacidad “operacional” a nuestros datalakes, soportando, por ejemplo, ACID transactions o Snapshots de nuestros datos, etc. Tenemos que comentar que no es incompatible con formatos columnares como Parquet u ORC, es más, son complementarios e incluso necesarios para Apache Iceberg, pero eso lo veremos más adelante.

Entre las principales características de Iceberg, encontramos las siguientes:

• ACID Compliance.
  Si has trabajado con Datalakes y debido a la naturaleza de las tecnologías que utilizan, no están especialmente diseñados para realizar continuas modificaciones sobre los datos ya persistidos. La filosofía de “escribe una vez, lee muchas veces” no siempre es posible. Cuando tenemos que realizar esas modificaciones en los datos, sin bien es cierto que son posibles hacerlas, hay que realizar procesos que pueden requerir borrar cierta información de manera temporal hasta que acaba el proceso de reescribir la información. Iceberg soporta de manera transparente operaciones update o delete a nivel de fila o en bloque (aunque veremos más adelante que hay que realizar algunas operaciones de mantenimiento) para el usuario.
• Snapshots.
  Iceberg nos permite tomar capturas del estado de una tabla en un momento dado. De esta manera, podemos consultar la tabla en un estado anterior. Esto nos permite hacer Time Traveling de los datos o estructura que teníamos en una tabla en un momento concreto. Podemos verlo como algo similar a los tag en un repositorio de git.
• Partition Evolution.
  En el diseño inicial de nuestra tabla, tendremos que tener especial cuidado en elegir bien nuestra estructura, tanto a nivel de tipos de datos como de columnas por las cuales vamos a particionar nuestros datos. Aún así, a medida que nuestra compañía crece (y con ella nuestros datos) es posible que necesitemos cambiar cómo particionamos esos datos. El particionamiento a nivel de año/mes puede haber sido una buena decisión en el pasado pero, a día de hoy, nos hemos dado cuenta que necesitamos más granularidad y el particionamiento tiene que ser año/mes/día. Esto es posible gracias a que Iceberg tiene guardada la especificación (Partition Spec) del particionamiento independientemente a dónde reside el dato y los cambios se aplican sobre esa spec. Iceberg usa hidden partitioning, así que no hay que escribir las consultas teniendo en cuenta la estructura que tiene la partición.
• Schema Evolution.
  Esta funcionalidad no es novedosa desde el punto de vista de Iceberg, ya que formatos como Parquet o Avro tienen esta capacidad y, como no puede ser de otra manera, Iceberg la soporta. Lo novedoso es que ofrece esta capacidad combinada con el Time Traveling y podemos hacer rollback, por ejemplo, a una versión anterior de una tabla.
• Compatibilidad.
  Dada la madurez que ya tienen ciertas tecnologías y lo implantadas que están en las arquitecturas datalake, es fundamental la compatibilidad de Iceberg con framework como Spark, Presto, Hive, etc.

¿Cómo funciona?

1. Capa de datos. En esta capa se almacena la información. Iceberg soporta diferentes tipos de formatos. Por defecto, utiliza Apache Parquet, pero se puede utilizar ORC o Avro.
2. Capa de metadatos. Gran parte de las capacidades que obtenemos al usar Iceberg son posibles gracias a esta capa.

• Ficheros de metadatos. Iceberg mantiene un lineage de los cambios producidos en la tabla. Cada cambio produce un nuevo fichero en formato JSON donde están todos los metadatos de la tabla.
• Ficheros de manifiesto. Son ficheros inmutables en formato avro que tienen una lista de todos los archivos de datos activos o borrados junto con los datos asociados a métricas, información de las particiones etc. Un conjunto de ficheros de manifiesto forman un manifest list, que es un snapshot de la tabla en un momento concreto del tiempo.

3. Catálogo

Iceberg utiliza el catálogo para agrupar por namespaces las tablas y, además, se encarga de recoger y llevar a cabo las operaciones de creación, renombrado o borrado de tablas.

Iceberg soporta la implementación de este catálogo automáticamente con:

• Servicio Rest.
• Hive Metastore.
• JDBC.
• Nessie.

Existen otros servicios de terceros que pueden actuar como catálogo de Iceberg. En nuestro caso, vamos a utilizar la implementación que ofrece Google a través de BigLake Metastore.

Para terminar este punto, podemos ver un diagrama que muestra cómo se relacionan los anteriores elementos en una tabla de Apache Iceberg.

💡Tip fundamental: mantenimiento de tablas:

Todas las capacidades que ofrece Iceberg son posibles gracias a su metadatado, y la premisa principal es que cada modificación de los datos genera un snapshot o versión de la tabla. Esta acumulación de snapshot, en modificaciones frecuentes sobre la tabla, provoca que se generen una cantidad significativa de ficheros pequeños o ficheros que ya no son referenciados. Para ello, Iceberg proporciona mecanismos out-of-box para lidiar con estas casuísticas.

• Compactación de ficheros de datos. Combina ficheros pequeños en ficheros más grandes.
• Expiración de Snapshots. Establecer un máximo tiempo de vida para snapshots que no están activos o que no han sido versionados.
• Reescritura de los manifest files.

Integrando Iceberg con BigQuery

Vamos a ver un ejemplo de cómo crear tablas de Iceberg, realizar operaciones de inserción y modificación de registros y consultarlas a través de BigQuery.

CREATE OR REPLACE PROCEDURE
 `data-platform-et-sandbox.procedure_dataset.iceberg_setup_3_3_v4`()
WITH CONNECTION `data-platform-et-sandbox.EU.sparkconnection` OPTIONS (engine='SPARK',
   runtime_version='2.0',
   properties=[("spark.sql.catalog.goodly.blms_catalog",
     "goodly"),
   ("spark.sql.catalog.goodly.gcp_project",
     "data-platform-et-sandbox"),
   ("spark.jars.packages",
     "org.apache.iceberg:iceberg-spark-runtime-3.3_2.12:1.2.0"),
   ("spark.sql.catalog.goodly.gcp_location",
     "EU"),
   ("spark.sql.catalog.goodly.warehouse",
     "gs://catalog_goodly_warehouse_bucket"),
   ("spark.sql.catalog.goodly",
     "org.apache.iceberg.spark.SparkCatalog"),
   ("spark.sql.catalog.goodly.catalog-impl",
     "org.apache.iceberg.gcp.biglake.BigLakeCatalog")],
   jar_uris=["gs://spark-lib/biglake/biglake-catalog-iceberg1.2.0-0.1.0-with-dependencies.jar"])
 LANGUAGE python AS R"""
from pyspark.sql import SparkSession

spark = SparkSession \
   .builder \
   .appName("BigLake Iceberg Example") \
   .enableHiveSupport() \
   .getOrCreate()

spark.sql("CREATE NAMESPACE IF NOT EXISTS goodly;")
spark.sql("CREATE DATABASE IF NOT EXISTS goodly.database;")
spark.sql("DROP TABLE IF EXISTS goodly.database.testing_iceberg_table_v4;")

spark.sql(
   "CREATE TABLE IF NOT EXISTS goodly.database.testing_iceberg_table_v4 (id bigint, demo_name string) USING iceberg TBLPROPERTIES(bq_table='database.testing_iceberg_table_v4', bq_connection='data-platform-et-sandbox.EU.cloudstorage_biglake_connection');")
""";

Una vez creado el procedimiento, lo ejecutamos y obtenemos una tabla externa de Iceberg en BigQuery.

CALL `data-platform-et-sandbox.procedure_dataset.iceberg_setup_3_3_v4`();

Cabe mencionar el atributo Source URI(s). No lo hemos comentado hasta ahora, pero hay un actor importante en la configuración: BigLake. Este es un motor de almacenamiento que se encarga de dar una visión única sobre diferentes formatos, servicios o incluso nubes públicas.

La tabla creada ya es consultable a través de SQL.

SELECT * FROM `data-platform-et-sandbox.database.testing_iceberg_table_v4` LIMIT 1000

Dado que es una tabla externa de BigQuery, no se pueden realizar consultas DML sobre ella. Para ello utilizaremos Spark, concretamente Dataproc, el servicio de Google Cloud que nos permite crear un cluster de Spark fácilmente.

Vamos a utilizar la shell de Spark para simular nuestro proceso con sentencias DML.

spark-sql --packages org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.5.2,org.apache.iceberg:iceberg-spark-extensions-3.5_2.12:1.5.2\
   --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions \
   --jars biglake-catalog-iceberg1.5.0-0.1.1-with-dependencies.jar \
   --conf spark.sql.catalog.goodly=org.apache.iceberg.spark.SparkCatalog \
   --conf spark.sql.catalog.goodly.catalog-impl=org.apache.iceberg.gcp.biglake.BigLakeCatalog \
   --conf spark.sql.catalog.goodly.gcp_project=data-platform-et-sandbox \
   --conf spark.sql.catalog.goodly.gcp_location=EU \
   --conf spark.sql.catalog.goodly.blms_catalog=goodly \
   --conf spark.sql.catalog.goodly.warehouse='gs://catalog_goodly_warehouse_bucket'

SELECT * from goodly.database.testing_iceberg_table_v4; 

spark-sql (default) > SELECT * from goodly.database.testing_iceberg_table_v4;
Time taken: 0.152 seconds

Insertamos algunos registros en la tabla:

insert into goodly.database.testing_iceberg_table_v4 values (1, "Ejemplo 1"), (2, "Ejemplo 2"), (3, "Ejemplo 3");

SELECT * from goodly.database.testing_iceberg_table_v4;

spark-sql (default) › SELECT * from goodly database.testing_iceberg_table_v4;
1       Ejemplo 1
2       Ejemplo 2
3       Ejemplo 3
Time taken: 0.779 seconds, Fetched 3 row (s)

Supongamos que queremos congelar esta primera versión para tenerla como referencia. Vamos a crear un branch llamado “primera-version” y lo mantendremos durante 7 días. Lo primero será conseguir la actual versión del snapshot de nuestra tabla:

SELECT * FROM goodly.database.testing_iceberg_table_v4.snapshots;

Con el id que obtenemos podemos generar un branch con los datos anteriormente mostrados.

ALTER TABLE goodly.database.testing_iceberg_table_v4 CREATE BRANCH `primera-version` AS OF VERSION 8510356156356767468 RETAIN 7 DAYS;

Realizamos una modificación en los datos y consultamos que se han realizado los cambios.

update goodly.database.testing_iceberg_table_v4 set demo_name="Modificación de Ejemplo 1" where id = 1;

SELECT * from goodly.database.testing_iceberg_table_v4 order by id asc;

spark-sql (default) > SELECT * from goodly. database.testing_iceberg_table_v4 order by id asc;
1       Modificación de Ejemplo 1
2       Ejemplo 2
3       Ejemplo 3

En este momento, tras realizar la modificación del registro con id=1, podemos consultar la versión “primera-versión” y podemos observar que no se muestran los cambios. Hemos realizado un time-traveling para consultar el estado de la tabla antes de ciertos cambios.

SELECT * FROM goodly.database.testing_iceberg_table_v4 VERSION AS OF 'primera-version';

spark-sql (default) > SELECT * FROM goodly. database.testing_iceberg_table_v4 VERSION AS OF 'primera-version';
1       Ejemplo 1
2       Ejemplo 2
3       Ejemplo 3

Por último, podemos ver algunas de las tareas de mantenimiento que hemos comentado anteriormente.

Para la tarea de compactación de ficheros de datos pr:

CALL goodly.system.rewrite_data_files(
 table => 'goodly.database.testing_iceberg_table_v4',
 strategy => 'binpack',
 options => map(
   'rewrite-job-order','bytes-asc',
   'target-file-size-bytes','1073741824', -- 1GB
   'max-file-group-size-bytes','10737418240' -- 10GB
 )
);

Expire snapshot:

CALL goodly.system.expire_snapshots(
 table => 'goodly.database.testing_iceberg_table_v4',
 older_than => TIMESTAMP '2024-07-24 11:00:00.000'
);

Conclusiones

Hemos visto cómo crear una tabla externa de bigquery en formato Iceberg y cómo poder realizar modificaciones sobre esta tabla con Spark, así como posibles opciones de mantenimiento. Para terminar, nos gustaría comentar 3 puntos.

Cuidado con el versionado. Utilizamos bastante tecnologías que tienen que encajar bien. El roadmap de tecnologías como Spark, Iceberg, Bigquery o BigLake son independientes. Si bien es cierto que se intenta integrar lo más rápidamente posible las nuevas versiones de unas tecnologías en otras, a veces es un dolor de cabeza por incompatibilidades. Para el ejemplo que hemos puesto, hemos tenido que tener en cuenta las siguientes versiones:

• Versión Scala.
• Versión Spark.
• Versión Iceberg
• Versión soportada por BigLake Catalog.
• Versión de Dataproc para el procesamiento de los datos (principalmente sentencias DML).

Hay que poner mucha atención a un posible upgrade de versión de uno de los componentes y ver si hay incompatibilidades.

¿Debo usar Iceberg o un formato de fichero columnar tipo Parquet y ORC?

Lo resumiremos en la siguiente tabla:

¿Merece la pena realizar una migración si ya tengo mi Datalake con Parquet?

Si tienes un Datalake que tiene pocas modificaciones de los datos y el tamaño medio de fichero es > 64-128 MB, quizás no merezca la pena realizar una migración. Pero, si vas a realizar nuevas ingestas o nuevas tablas, quizás puede tener sentido ir introduciendo Iceberg poco a poco.