30 de enero de 2025

De BigQuery a Lakehouse: Cómo creamos una plataforma de análisis de datos a escala de petabytes - Parte 1

En TRM Labsproporcionamos herramientas Inteligencia en Blockchain para ayudar a instituciones financieras, criptoempresas y agencias gubernamentales a detectar e investigar delitos y fraudes financieros relacionados con las criptomonedas.

Nuestra plataforma de análisis procesa petabytes de datos de cadenas de bloques en más de 30 cadenas de bloques y responde a más de 500 consultas de clientes por minuto con una latencia ultrabaja, gracias a Postgres distribuido y BigQuery. Durante años, hemos optimizado BigQuery para escalarlo de forma eficiente, pero cuando tuvimos que ejecutarlo in situ, nos topamos con un muro. BigQuery no podía satisfacer nuestras necesidades multientorno, y escalar Postgres para la ingestión masiva de datos resultaba demasiado costoso. Necesitábamos una solución abierta, autoalojada, segura y de alto rendimiento, así que creamos un lago de datos a escala de petabytes con Apache Iceberg y StarRocks para satisfacer nuestras necesidades analíticas de cara al usuario. Así es como lo hicimos, lo que aprendimos y por qué podría cambiar su forma de pensar sobre la arquitectura de datos.

1. Plataforma de datos de primera generación

En nuestra Plataforma de Datos de Primera Generación, utilizamos un clúster Postgres distribuido(Citus Data) para búsquedas puntuales rápidas y pequeñas preagregaciones. Cuando las cargas de trabajo superaban la capacidad del clúster Postgres distribuido, federábamos las consultas de mayor tamaño y las agregaciones ad hoc a BigQuery.

La [Figura 1] muestra cómo la plataforma de datos de primera generación de TRM gestionaba los análisis orientados al usuario y dirigía las consultas a través de Postgres y BigQuery.

2. Más allá de BigQuery y hacia una nueva generación de lagos de datos abiertos

Aunque BigQuery satisfizo bien las necesidades analíticas de nuestros clientes durante años, nos encontramos con limitaciones a medida que nos expandíamos a despliegues multientorno, incluidos los entornos locales. La necesidad de compartir datos Analítica de blockchain en varios sitios hizo que los servicios gestionados como BigQuery fueran poco prácticos, y nuestras cargas de trabajo de servicio requerían un nuevo enfoque de escalado.

Principales requisitos que impulsan el cambio

Despliegue en varios sitios: La necesidad de desplegar nuestra plataforma en varios sitios locales, manteniendo al mismo tiempo la capacidad de compartir datos, requería el uso de soluciones de código abierto que pudieran desplegarse en Kubernetes.
Escala y rendimiento: Nuestra mayor carga de trabajo de cara al cliente, que anteriormente dependía de BigQuery para el servicio, contiene más de 115 TB de datos y crece entre un 2 % y un 3 % mensualmente. Estas consultas de lectura implican complejas uniones multinivel con filtros basados en el tiempo y en matrices. Alcanzar una latencia P95 de tres segundos con una alta concurrencia con BigQuery resultó un reto sin invertir en costosas ranuras de cálculo. Trasladar cargas de trabajo de este tipo a Postgres distribuido resultaría caro sólo por razones de almacenamiento.

Nuestra plataforma de datos de nueva generación debía combinar la flexibilidad de un lago de datos con el rendimiento y la fiabilidad de un almacén de datos. Construir un lago de datos moderno en torno a Apache Iceberg permitía la interoperabilidad con motores de consulta y motores de computación distribuida compatibles con la especificación Iceberg. Tras evaluar varios motores de consulta, elegimos StarRocks. Esta combinación de Apache Iceberg y StarRocks satisfacía nuestros requisitos de despliegue y rendimiento en varios sitios, al tiempo que proporcionaba ventajas clave para el crecimiento futuro.

Oportunidades que vimos

Estándares abiertos: La implementación de código abierto de Apache Iceberg proporciona evolución de esquemas, viajes en el tiempo y gestión eficiente de metadatos en almacenamiento de objetos. Su flexibilidad permite el despliegue en entornos locales multisede, lo que la hace perfecta para compartir datos Analítica de blockchain en múltiples ubicaciones.
Lago de datos de alto rendimiento: StarRocks proporciona una latencia ultrabaja y una alta concurrencia a través de un almacenamiento en caché avanzado y un procesamiento de consultas totalmente vectorizado. La combinación de StarRocks con Iceberg nos proporciona el rendimiento de un almacén de datos al tiempo que mantiene la flexibilidad del lago de datos.
Independencia del motor de consulta: Construir nuestro lago de datos sobre Apache Iceberg nos da flexibilidad para integrar cualquier motor de consulta compatible, lo que nos permite adaptarnos a la evolución de la tecnología. En el año transcurrido desde que realizamos nuestras pruebas comparativas, hemos observado un rápido avance en el rendimiento de los motores de consulta. Estamos impacientes por reevaluar tanto las soluciones ya existentes (por ejemplo, Trino y DuckDB) como las recién llegadas (por ejemplo, Clickhouse y Crunchy Data Warehouse). Esta flexibilidad nos mantiene a la vanguardia del rendimiento y la rentabilidad, independientemente de un único proveedor.
Reducción de costes: Dado que los datos y metadatos se almacenan de forma eficiente en el almacenamiento de objetos, identificamos una oportunidad para migrar las cargas de trabajo de nuestro clúster Postgres distribuido, reduciendo los costes de almacenamiento SSD.

3. Por qué Apache Iceberg + StarRocks para un lago de datos

Dado que las implantaciones en múltiples entornos, incluido el local, se están convirtiendo en un requisito clave, necesitábamos una solución alternativa para los clientes que se enfrentan a casos de uso analítico. Nuestro trabajo con BigQuery y Postgres nos permitió hacer algunas observaciones clave:

Minimizar los datos leídos en el momento de la consulta es fundamental mediante el uso de la compresión de datos, la agrupación y la partición para optimizar las exploraciones.
Los índices tradicionales de tipo árbol B resultan ineficaces a escala de petabytes.
La moderna ejecución vectorial de la CPU (por ejemplo, SIMD) acelera considerablemente el procesamiento de las consultas.
El escalado horizontal permite una gran concurrencia manteniendo unos costes razonables.
Separar la computación del almacenamiento nos permite cambiar con flexibilidad entre motores de consulta o combinarlos para obtener un rendimiento óptimo de la carga de trabajo, sin duplicar datos.

A partir de esta información, dejamos atrás los almacenes de datos OLAP tradicionales (por ejemplo, Clickhouse) y empezamos a explorar el mercado emergente de los "lagos de datos". Necesitábamos tomar dos decisiones clave: (1) nuestro formato de almacenamiento y (2) nuestro motor de consulta.

3.1 Formato de almacenamiento

En TRM, nuestras necesidades de almacenamiento, especialmente con la llegada de cadenas de bloques de alto rendimiento, crecen exponencialmente cada año. Necesitábamos asegurarnos de que nuestro sistema de almacenamiento fuera a la vez eficiente y rentable a medida que incorporábamos nuevas cadenas de bloques en el futuro.

Empezando por el coste, sabíamos que teníamos que abandonar las unidades SSD y pasar a los almacenes de objetos, ya que incluso el almacén de objetos más caro es 4 veces más barato que la unidad SSD más barata.

Con nuestro conjunto de opciones reducido a los almacenes de objetos, evaluamos 3 de los formatos de almacenamiento más populares para construir un lago de datos.

Aunque Delta Lake ofrecía características y rendimiento convincentes, lo descartamos debido a su falta de evolución de particiones y a su solapamiento con Iceberg para el análisis a gran escala y el procesamiento por lotes. A continuación, realizamos una prueba comparativa con Apache Hudi; nuestra tabla Hudi de mejor rendimiento era 3 veces más lenta que Apache Iceberg.

Apostamos por Apache Iceberg, que ofrecía un rendimiento de lectura excepcional al tiempo que contaba con una amplia adopción por parte de la comunidad, una comunidad de desarrollo activa y una amplia compatibilidad con catálogos y motores de consulta.

3.2 Motor de consulta

Una vez elegido nuestro formato de tabla, evaluamos varios motores de consulta compatibles con Iceberg en sus versiones de código abierto. Evaluamos tres motores: Trino, StarRocks y DuckDB. Nuestras pruebas mostraron que StarRocks superaba sistemáticamente a los demás (véase la figura 2).

Trino: Motor de consulta distribuido de código abierto diseñado para consultar conjuntos de datos muy grandes.
StarRocks: Un motor de consulta rápido y de código abierto para análisis dentro y fuera del lago de datos.
DuckDB: Motor de consulta SQL analítica en proceso de código abierto.

[Figura 2] Los resultados de nuestro benchmark muestran comparaciones de rendimiento de consultas para operaciones de búsqueda/filtrado en múltiples configuraciones de tres motores, probadas en un conjunto de datos de 2,57 TB de Analítica de blockchain . StarRocks ofreció tiempos de respuesta sistemáticamente superiores en todas las configuraciones.

3.3 Resultados de la experimentación

Nuestra experimentación se centró en dos cargas de trabajo principales [6.1.2]: consultas de búsqueda puntual con filtrado y consultas de agregación compleja con filtrado. Utilizamos JMeter para realizar pruebas de carga y verificar que los motores de consulta podían mantener el rendimiento en condiciones de alta concurrencia.

3.3.1 Experimento con Lookup/Filter

La figura 2 muestra los resultados de esta carga de trabajo, en la que probamos consultas de búsqueda por puntos y por rangos que devolvían pequeños subconjuntos de un conjunto de datos de 2,57 TB. Observamos lo siguiente:

StarRocks: Consiguió el mejor rendimiento en todas las configuraciones, con tiempos de respuesta de consulta tan bajos como 470 ms con almacenamiento de datos en caché.
Trino: Tiempos de respuesta entre 1.410 ms y 1.030 ms, que varían con el tamaño del clúster.
DuckDB: alcanzó un rendimiento razonable de 2-3 segundos en un único nodo potente. Detuvimos las pruebas de DuckDB después de este benchmark debido a limitaciones en su soporte de tablas Iceberg. Estamos esperando a que la extensión Iceberg de DuckDB añada soporte para predicate pushdown para una futura evaluación.

3.3.2 Experimento de agregación compleja

[Figura 3] Nuestros resultados de referencia comparan el rendimiento de las consultas de agregación compleja entre Trino y StarRocks en diferentes configuraciones de clúster.

En nuestro siguiente experimento, probamos consultas que realizaban operaciones SUM, COUNT y GROUP BY con filtros de matriz y rango de fechas en un conjunto de datos de 2,85 TB. Nuestros resultados fueron los siguientes:

StarRocks: StarRocks manejó cargas de trabajo agregadas complejas excepcionalmente bien, logrando latencias de ~2 segundos sin almacenamiento en caché y tan bajas como 500 ms con almacenamiento en caché en nuestro clúster de prueba más grande.
Trino: Aunque el rendimiento de Trino mejoró significativamente con clusters más grandes, alcanzó un techo de rendimiento en aproximadamente 2,5 segundos.

3.3.3 Pruebas de resistencia

Hemos utilizado JMeter para probar el rendimiento de Trino y StarRock en condiciones de alta concurrencia.

StarRocks: StarRocks superó sistemáticamente a Trino durante las pruebas de alta concurrencia, tanto para cargas de trabajo de búsqueda como de agregación. Cuando se habilitó el almacenamiento de datos en caché, el rendimiento mejoró aún más.
Trino: El rendimiento de Trino se degradaba a medida que aumentaba la carga de usuarios concurrentes. Cuando realizamos estas pruebas a principios de 2024, Trino no disponía de funciones de almacenamiento en caché de datos para las tablas del lago de datos. Aunque esta función se añadió posteriormente en Trino 439, aún no la hemos evaluado.

[Figura 4] Resultados de una prueba multihilo utilizando la carga de trabajo de agregaciones complejas (izquierda) y la carga de trabajo de búsqueda/filtrado (derecha).

4. Nuestro camino a seguir

[Figura 5] Ilustra nuestra Plataforma de Datos de Próxima Generación para análisis de cara al usuario.

Sobre la base de nuestra evaluación de tres formatos de tabla abiertos y la experimentación con múltiples motores de consulta, decidimos construir un lago de datos con StarRocks y Apache Iceberg como componentes centrales para abordar los requisitos clave para la construcción de la plataforma de datos de TRM a través de múltiples sitios y para mejorar el rendimiento para nuestros clientes.

Data Lakehouse ofrece una doble ventaja: Nuestro enfoque de data lakehouse combina la flexibilidad de un lago de datos con el rendimiento de un almacén de datos, lo que permite realizar análisis rápidos y fiables de cara al cliente.
Apache Iceberg: Con sus estándares abiertos, su sólida evolución de esquemas y su eficaz gestión de metadatos, Iceberg ofrece la interoperabilidad entre motores que necesitamos.
StarRocks: Gracias a la optimización estratégica de la partición de tablas Iceberg, la agrupación en clústeres, el dimensionamiento de clústeres StarRocks y las estrategias de almacenamiento en caché, conseguimos un rendimiento excepcional con baja latencia y alta concurrencia. Estas mejoras se tradujeron en una mejora del 50% en los tiempos de respuesta P95 y una reducción del 54% en los errores de tiempo de espera de las consultas, lo que garantizó que pudiéramos cumplir nuestros objetivos de rendimiento de las consultas.
Las pruebas son la clave: Las cargas de trabajo del mundo real revelaron patrones de uso y oportunidades de optimización que los puntos de referencia por sí solos no podían identificar, lo que subraya la importancia crítica de realizar pruebas exhaustivas a escala.

En la segunda parte de esta serie, exploraremos cómo hemos dado vida a esta arquitectura, desde el despliegue de Apache Iceberg en el almacenamiento de objetos hasta la optimización de StarRocks para despliegues en múltiples entornos, incluidos los entornos locales.

5. Ingeniería de datos en TRM

En TRM Labsnos mueve una misión audaz: proteger a la civilización de los delitos de IA y construir un mundo más seguro para miles de millones de personas. Mediante el avance de Inteligencia en Blockchain y la creación de la plataforma de datos blockchain del futuro, abordamos los retos más difíciles de la delincuencia financiera y la Analítica de blockchain.

Nuestra misión está impulsada por expertos como:

Vijay Shekhawat (coautor), miembro clave de los TRM Labs aporta una gran experiencia en streaming en tiempo real, arquitecturas Data Lakehouse y creación de canalizaciones seguras y de alto rendimiento para análisis a escala de petabytes, impulsando la misión de TRM.
Andrew Fisher (coautor), ingeniero de software de los TRM Labs especializado en cargas de datos por lotes a gran escala y soluciones Data Lakehouse que impulsan el análisis a escala de petabytes en la lucha contra el fraude de criptomonedas.
Elena Tomlinson, Moamen Ali, Brice Kamgne, Steven Hope y Sharad Bhadouria, colaboradores clave de los TRM Labs han desempeñado un papel decisivo en la validación de la arquitectura del lago de datos mediante la migración de cargas de trabajo clave. Su trabajo ha sido fundamental para garantizar la escalabilidad, la eficiencia y el alto rendimiento de la plataforma en casos de uso reales.

Agradecimientos especiales a Michael Andrews y Amit Plaha por sus perspicaces revisiones y su constante dedicación a la excelencia técnica a lo largo de este proyecto.

Únete a nuestro equipo

Nuestros ingenieros están construyendo un lago de datos a escala de petabytes con tiempos de respuesta de latencia ultrabaja, abordando algunos de los retos más difíciles en Analítica de blockchain para luchar contra la delincuencia y construir un mundo más seguro para miles de millones de personas. ¿Emocionado y listo para dejar huella? ¿O crees que puedes hacerlo mejor? Explore las oportunidades y presente su candidatura hoy mismo .

‍

6. Apéndice / experimentos detallados

6.1 - Viaje de experimentación

Para evaluar el rendimiento de las consultas de Apache Iceberg, realizamos pruebas comparativas de cargas de trabajo de lectura típicas utilizando diferentes motores de consulta para encontrar el que mejor cumplía nuestros requisitos de rendimiento y escala. Nos centramos en dos categorías de cargas de trabajo: búsqueda/filtro y agregación compleja.

6.1.1 - Preparación de los datos

Paracada carga de trabajo, creamos tablas Iceberg a partir de nuestros datos existentes de BigQuery. Nuestras consultas suelen hacer referencia a direcciones específicas de cadenas de bloques, entidades fuera de la cadena o periodos de tiempo, por lo que dividimos estratégicamente nuestros conjuntos de datos para garantizar que las consultas solo accedieran a subconjuntos de datos relevantes. Además:

Exportamos múltiples tablas (2-3 TBs) de BigQuery a formato Parquet en Google Cloud Storage.
Utilizando PySpark, transformamos estos archivos Parquet en tablas Iceberg, optimizándolas con configuraciones de bucketing y ordenación.
Utilizamos Dataproc Metastore para mantener los metadatos de esquemas y tablas.

6.1.2 - Descripción de la carga de trabajo

6.1.2.1 Experimento con búsqueda/filtro

-- Ejemplo de tabla Iceberg para transacciones blockchain
CREATE TABLE blockchain_transactions (
    transaction_id STRING,
    block_number BIGINT,
    from_address STRING,
    to_address STRING,
    amount DECIMAL(38,18), 
    timestamp TIMESTAMP,
    cadena STRING
) USING iceberg
PARTITIONED BY (bucket(transaction_id, 3000));

-- Ejemplo de consulta para filtrar transacciones por ID
SELECT 
    transaction_id,
    dirección_de,
    a_dirección,
    importe
    marca de tiempo
FROM blockchain_transactions 
WHERE transaction_id = '0x1234abcd...';

6.1.2.2 Experimento de agregación compleja

-- Ejemplo de consulta para agregación compleja
SELECT
    e.id::TEXT AS entity_id,
    SUM(CASE WHEN t.transaction_type = 'type_a' THEN t.weight_normalized END) AS type_a_volume,
    SUM(CASE WHEN t.transaction_type = 'type_b' THEN t.weight_normalized END) AS type_b_volume, 
    SUM(t.peso_normalizado) COMO volumen_total,
    SUM(CASE WHEN t.transaction_type = 'type_a' THEN 1 ELSE 0 END) AS type_a_count,
    SUM(CASE WHEN t.transaction_type = 'type_b' THEN 1 ELSE 0 END) AS type_b_count, 
    COUNT(*) COMO total_count
FROM transacciones t
JOIN entidades e ON t.entidad_id = e.id
JOIN categorías c ON e.category_id = c.id
WHERE t.cadena = __cadena::TEXTO
  AND t.chain_id = __chain_id::TEXT
  AND t.timestamp BETWEEN __start_date::TIMESTAMP AND __end_date::TIMESTAMP
  AND t.path_length BETWEEN __min_hop AND __max_hop
  Y (
        __entity_ids IS NULL 
        OR t.entity_id = ANY(__entity_ids)
      )
  AND (
        __category_ids IS NULL 
        OR EXISTS (
            SELECT 1 
 FROM unnest(t.entity_category_ids) AS cat_id
            WHERE cat_id = ANY(__category_ids)
        )
      )
  AND e.id IS NOT NULL
GROUP BY e.id
ORDER BY __sort_column __sort_order NULLS LAST
LIMIT __limit
OFFSET __offset;

6.1.3 - Infraestructuras

‍Paraevaluar el escalado del rendimiento, desplegamos cada motor de consulta con distintas configuraciones de recursos informáticos. En la siguiente sección se detallan las especificaciones de configuración.

StarRocks: Desplegado en GKE utilizando el operador StarRocks Kubernetes.
Trino: Desplegado en Google Dataproc (plataforma Hadoop/Spark gestionada).
DuckDB: Desplegado en máquinas virtuales GCP.

6.2 - Infraestructuras

6.2.1 Trino

Cómo: Desplegado en Google Dataproc (plataforma Hadoop/Spark gestionada) por su sencillez y sus capacidades de autoescalado integradas.
Versión: Trino 433
Tamaño de los grupos para la experimentación:

6.2.2 StarRocks

Cómo: Desplegado en GKE utilizando el operador StarRocks Kubernetes.
Versión: 3.1
Tamaño de los grupos para la experimentación:

6.2.3 DuckDB

Cómo: Desplegado en máquinas virtuales de Google Cloud Platform (GCP).
Versión: 0.9.2
Tamaño de los grupos para la experimentación:

6.3 - Metodología de las pruebas de resistencia

Herramientas y configuración:

JMeter: Se utiliza para simular las peticiones de los usuarios al servidor de consultas.
- JMeter genera cargas de servidor realistas simulando peticiones simultáneas de usuarios. Admite múltiples protocolos, incluidos JDBC, HTTP y FTP.
- Componentes clave:
  - Grupo de hilos: Conjunto de usuarios simulados que realizan acciones especificadas, con recuento de usuarios, periodo de aceleración y frecuencia de ejecución ajustables.
Consultas parametrizadas y datos CSV: Las consultas utilizan parámetros dinámicos de archivos CSV para crear cargas de trabajo diversas y realistas.
Hilos concurrentes JMeter: 1, 3, 5, 8, 13, 21, 34, 55 (simula usuarios concurrentes)
Ejecución:
- Duración: Cada prueba dura 1 minuto continuo.
- Límite de consulta: las pruebas funcionan sin límites de consulta para medir la capacidad máxima del sistema.