Data Marts de Delhivery: viaje de migración de OLTP a HTAP

Como plataforma de cumplimiento líder para el comercio digital en la India, Delhivery procesa un millón de paquetes al día, los 365 días del año. Sus 24 centros de clasificación automatizados, 101 centros, más de 3100 centros de entrega directa, más de 1000 centros asociados, más de 11 000 flotas y más de 60 000 miembros del equipo funcionan sin problemas gracias a una amplia red de dispositivos IoT. Miles de eventos de datos y mensajes entran y salen de nuestras tuberías cada segundo. Esto equivale a un enorme volumen de datos diario en terabytes, lo que hace que la visibilidad operativa sea crucial para nosotros y nuestras partes interesadas.

Al reconocer los requisitos, decidimos crear data marts: bases de datos centralizadas y eventualmente consistentes que ofrecen a los usuarios acceso rápido a datos comerciales preagregados. Esto permite a nuestras partes interesadas acceder rápidamente a información empresarial sin tener que buscar en todo un almacén de datos.

Sin embargo, con esta escala desalentadora, uno de los principales desafíos fue mantener la integridad de los datos y la baja latencia al mismo tiempo que se proporcionaba la capacidad para cargas de trabajo analíticas.

En este blog, voy a analizar todas mis ideas mientras migraba nuestros data marts de Amazon Aurora a TiDB, una base de datos SQL distribuida de procesamiento híbrido transaccional/analítico (HTAP). Con suerte, esta publicación puede brindar información a los líderes de ingeniería de datos, administradores de bases de datos o arquitectos de datos que estén considerando una migración similar a TiDB o cualquier otra base de datos HTAP.

OLTP, OLAP y HTAP

Para comprender mejor el caso de los data marts en tiempo real en Delhivery, primero familiaricémonos con tres conceptos que son el núcleo de nuestro caso de uso: OLTP, OLAP y HTAP:

• OLTP: Los sistemas de procesamiento de transacciones en línea (OLTP) están diseñados para aplicaciones orientadas a transacciones, garantizando la integridad de los datos a través de propiedades ACID (Atomicidad, Consistencia, Aislamiento, Durabilidad).
• OLAP: Los sistemas de procesamiento analítico en línea (OLAP) permiten el análisis multidimensional de alta velocidad de grandes volúmenes de datos, lo que ayuda en la toma de decisiones basada en datos.
• HTAP: El procesamiento híbrido de transacciones/analítico (HTAP) combina funcionalidades OLTP y OLAP, lo que permite análisis en tiempo real de datos transaccionales.

Caso de uso de Data Marts en tiempo real en Delhivery

Los data marts en tiempo real se diferencian de los data marts tradicionales en que ingieren datos en tiempo real, no en intervalos específicos. Estos mercados de datos son fundamentales para la toma de decisiones operativas terrestres en Delhivery porque no podemos permitirnos ningún retraso en la sincronización de estos eventos.

Nuestro viaje al mercado de datos en tiempo real comenzó en 2020 cuando identificamos la necesidad de paneles centralizados, específicamente el panel EYE. El objetivo de este panel era proporcionar visibilidad operativa en tiempo real de las operaciones terrestres, permitiendo la toma de decisiones basada en datos actualizados. Ejemplos de usos incluyen:

• Planificación y visibilidad de vehículos: monitoreo en tiempo real de los horarios de conexiones entrantes y salientes para los centros Delhivery.
• Seguimiento del desempeño: Seguimiento continuo del desempeño de las instalaciones de Delhivery.
• Visibilidad de control centralizado: Proporcionar al equipo central información precisa sobre los bloqueadores terrestres para tomar las acciones adecuadas. Esto podría deberse a varios factores, como una caída en el rendimiento del centro, antigüedad de los envíos o congestión en las conexiones entrantes y salientes.
• Cumplimiento: Seguimiento de métricas de cumplimiento de opciones de compra y venta

Implementación inicial y los desafíos

Pensamos en resolver nuestros casos de uso utilizando herramientas de almacenamiento de datos como Redshift y Snowflake, pero ninguna de estas soluciones funcionó para nosotros considerando el patrón de diseño y los requisitos para la ingesta de datos en tiempo real junto con la fusión.

Por lo tanto, inicialmente elegimos Aurora (PostgreSQL) para nuestro caso de uso de data mart.

El proceso de ingesta de datos en torno a Aurora

Diseñamos nuestros data marts en tiempo real utilizando Spark Streaming y Aurora. Nuestro proceso de vaporización fue muy simple: leer datos de Kafka, procesar datos en microlotes Spark y realizar operaciones de inserción en Aurora.

Nuestra base de datos se modeló utilizando una arquitectura de múltiples capas, que consta de una capa sin procesar, una capa particionada y una capa de data marts. Los usuarios no tenían acceso para ver o modificar datos en la capa sin procesar. La capa particionada se mantiene para mantener todas las tablas particionadas (tablas de dimensiones en general). A continuación se muestra un diseño de esquema simple de nuestra base de datos:

Desafíos que enfrentamos con Aurora

Inicialmente, el sistema funcionó bien, hasta que tuvo que manejar un rendimiento superior a 3.000 mensajes por segundo. Esto marcó el inicio de varios desafíos:

• Limitación de escalabilidad: a medida que excedimos un rendimiento de 3 000 mensajes por segundo, las limitaciones de las operaciones de entrada/salida por segundo (IOPS) de Aurora se convirtieron en un cuello de botella. La limitación de escalabilidad había comenzado a afectar nuestras operaciones.

  
  

• Problema de hinchazón: cada actualización de registro conducía a la creación de un nuevo registro y una tupla muerta (versión anterior del registro). Cuando la tasa de producción de estas tuplas muertas superó el proceso de limpieza, se produjo una hinchazón. Dado que VACUUM FULL no pudo reclamar el almacenamiento, el uso del disco aumentó continuamente. Para aproximadamente 5 TB de datos, Aurora usaba más de 30 TB de almacenamiento.

  
  

• Carga de mantenimiento: El problema de la hinchazón está directamente relacionado con nuestros desafíos de mantenimiento. Con más de 70 canalizaciones y un QPS de escritura total que supera los 5.000 mensajes/segundo, descubrimos que el proceso de limpieza automática de PostgreSQL, Auto Vacuum, no logró seguir el ritmo de la tasa de generación de tuplas muertas. Por lo tanto, es necesario ejecutar VACUUM o VACUUM FULL manualmente para recuperar la base de datos. Nuestros intentos con herramientas PostgreSQL como pg_repack y pgcompacttable tampoco tuvieron éxito. En consecuencia, el mantenimiento se volvió cada vez más complejo y consumía más tiempo.

• Costo: para acomodar la carga de trabajo de lectura y escritura, tuvimos que escalar a los nodos más altos disponibles (24XLarge). Esto generó un gasto de aproximadamente $100 000 por mes para un clúster Aurora de tres nodos. Con esta escala, Aurora resultó ser costosa debido al escalado automático de IOPS.

Buscando alternativas

Para resolver las limitaciones de Aurora, nos propusimos encontrar una alternativa mejor que cumpliera con los siguientes requisitos:

• Escalable con QPS de escritura alto: la base de datos debe admitir al menos 10 000 QPS de escritura y ser escalable horizontalmente.
• Análisis en tiempo real: la base de datos debe poder proporcionar capacidades OLAP de alta velocidad o en tiempo real.
• Totalmente distribuida: la base de datos debe distribuirse en varios sitios para proporcionar alta disponibilidad y tolerancia a fallas.
• Gran coherencia: la base de datos debe mantener una gran coherencia, garantizando que todos los usuarios vean los mismos datos.

Teniendo en cuenta todos los requisitos anteriores, inicialmente exploramos muchas alternativas de PostgreSQL, incluidas Spanner y Yugabyte, porque queríamos mantener nuestra gestión de cambios al mínimo.

Llave

Spanner es un servicio distribuido de almacenamiento y gestión de bases de datos SQL ofrecido por Google. Está completamente administrado en Google Cloud Platform (GCP). Sin embargo, descubrimos que Spanner podría no ser un buen caso de uso para nuestra arquitectura debido a las siguientes razones:

• Spanner no admite esquemas.
• No encontramos las herramientas adecuadas para cargar datos históricos. Exploramos Harbourbridge, una herramienta de código abierto para la evaluación y migración de Spanner. Sin embargo, tenía limitaciones en torno a los 100 GB de carga de datos.

Yugabyte

YugabyteDB es una base de datos SQL distribuida transaccional de alto rendimiento para aplicaciones nativas de la nube, desarrollada por Yugabyte. Esta base de datos está muy cerca de nuestro caso de uso porque era totalmente compatible con PostgreSQL, escalable horizontalmente y completamente distribuida. Desafortunadamente, no funcionó tan bien debido a su limitación de escalabilidad. Nuestros criterios de éxito exigían más de 7.000 transacciones por segundo, pero Yugabyte solo pudo escalar hasta 5.000.

También analizamos otros posibles candidatos como BigQuery, pero ninguno de ellos cumplió bien con nuestros requisitos.

Aterrizando con TiDB

Después de las alternativas de PostgreSQL anteriores, decidimos agregar HTAP a nuestros requisitos, lo que nos llevó a TiDB. Admite escalabilidad, coherencia, disponibilidad, topología de implementación multisitio y muchas más funciones listas para usar. Como base de datos distribuida, TiDB tiene múltiples componentes que se comunican entre sí y forman un sistema TiDB completo.

• TiDB: es el componente de procesamiento de SQL sin estado que proporciona al usuario el punto final de cara al cliente. Localiza el nodo TiKV correcto para conectarse desde PD para obtener los datos.
• TiKV: es un almacén de datos transaccional distribuido de valores clave que mantiene los datos en el rango izquierda-cerrada-derecha-abierta. Los datos se guardan en fragmentos con múltiples réplicas. TiKV utiliza el protocolo Raft para la replicación.
• PD: el controlador de ubicación (PD) mantiene los metadatos del clúster, como las ubicaciones de las réplicas de fragmentos, y también es responsable de programar los fragmentos en los nodos TiKV. El líder de PD maneja dichas tareas mientras otros nodos mantienen una alta disponibilidad.
• TiFlash: la extensión de almacenamiento en columnas que utiliza el protocolo Multi-Raft Learner para replicar datos de TiKV en tiempo real, lo que garantiza datos consistentes entre el motor de almacenamiento basado en filas de TiKV.

Las siguientes características de TiDB abordaron nuestros desafíos clave y cumplieron con nuestros requisitos operativos:

• Fácil escalado

  El diseño de la arquitectura TiDB separa la informática del almacenamiento, lo que le permite ampliar o ampliar la capacidad informática o de almacenamiento en línea según sea necesario. El proceso de escalado es transparente para el personal de operaciones y mantenimiento de aplicaciones.

• Cumple con ÁCIDO

  TiDB es compatible con MySQL y admite transacciones listas para usar. Admite tipos de transacciones tanto optimistas como pesimistas. Esto la hace única respecto a otras bases de datos.

• Altamente disponible

  TiKV almacena datos en múltiples réplicas y utiliza el protocolo Multi-Raft para obtener el registro de transacciones. Sólo se puede confirmar una transacción cuando los datos se han escrito correctamente en la mayoría de las réplicas. Esto garantiza una gran coherencia y alta disponibilidad cuando una minoría de réplicas deja de funcionar.

• HTAP en tiempo real

  TiDB combina almacenamiento en filas (TiKV) y almacenamiento en columnas (TiFlash) en la misma arquitectura, formando una pila tecnológica optimizada que facilita la producción de análisis en tiempo real de datos operativos.

Nuestra infraestructura TiDB

Nuestra infraestructura TiDB se implementa en las máquinas virtuales de los principales proveedores de servicios en la nube. Usamos TiUP, el administrador de paquetes de TiDB, para administrar el clúster y todas las operaciones administrativas. Nuestro clúster se implementa en 3 zonas disponibles (AZ).

Nuestras configuraciones de clúster son las siguientes:

• PD: la capa PD tiene 3 nodos divididos en Multi-AZ. El líder de PD maneja dichas tareas mientras otros nodos mantienen una alta disponibilidad.
• TiDB: La capa TiDB tiene 9 nodos de la familia n2-highmem-8. Estos nodos se eligieron en función de los requisitos de memoria, con 64 GB de RAM y CPU de 8 núcleos asignados para cada nodo TiDB.
• TiKV: la capa TiKV tiene 15 nodos de la familia n2-highmem-16 que tiene 128 GB de RAM y 16 CPU vCORE.

Al implementar nuestro clúster TiDB en múltiples AZ y seleccionar cuidadosamente los tipos de nodos para satisfacer nuestras necesidades de procesamiento y memoria, hemos creado una infraestructura sólida y de alta disponibilidad capaz de manejar nuestros requisitos de alto rendimiento de datos.

Ajustando TiDB para nuestro caso

Para que funcione en nuestro caso de uso, trabajamos estrechamente con el equipo de PingCAP para ajustar la base de datos. Estos son algunos de los ajustes críticos que hicimos:

Optimización del índice

Establezca los siguientes parámetros antes de iniciar el índice.

 SET @@global.tidb_ddl_reorg_worker_cnt = 16; SET @@global.tidb_ddl_reorg_batch_size = 4096;

Restablecer los valores predeterminados después de la creación del índice.

 SET @@global.tidb_ddl_reorg_worker_cnt = 4; SET @@global.tidb_ddl_reorg_batch_size = 256;

Poda de partición

Esto es principalmente importante para tablas particionadas. Analiza las condiciones del filtro en las declaraciones de consulta y elimina (poda) las particiones cuando no contienen ningún dato requerido.

 SET @@session.tidb_partition_prune_mode = 'dynamic';

Análisis de sintonización

A veces, el analizador automático de TiDB falla si se ingiere un gran volumen de datos. En ese caso, todas las consultas podrían utilizar el plan de ejecución incorrecto y terminar escaneando la tabla completa. Para evitar tal situación, realizamos los siguientes cambios en las configuraciones de TiDB:

 set global tidb_max_auto_analyze_time = 86400; set global tidb_enable_pseudo_for_outdated_stats = off; set global tidb_sysproc_scan_concurrency = 15;

Si está trabajando con tablas particionadas, le sugerimos que ejecute las operaciones de análisis de tablas manualmente para una partición a la vez para evitar fallas en el análisis.

A través de ajustes como estos, pudimos optimizar de manera efectiva nuestro uso de TiDB, de modo que podamos lograr un rendimiento óptimo para nuestro mercado de datos en tiempo real.

Nuestra experiencia con TiDB

• Rendimiento de consultas mejorado

  Hemos evaluado más de 400 consultas y descubrimos que todas las consultas se ejecutan dentro de SLA. Incluso hemos visto una ganancia de rendimiento del 15 al 20 % en las consultas P95.

• Fácil migración

  Usamos la herramienta TiDB Lighting para migrar todos los datos históricos de nuestra tabla de Postgres a TiDB. Esta herramienta es muy fácil de usar y muy rápida. Pudimos cargar terabytes de datos en aproximadamente 2 o 3 horas. Sin embargo, vale la pena señalar que se requieren muchos ajustes antes de cargar datos tan grandes.

• Apoyo fuerte

  Pasamos por un par de contratiempos durante la configuración de la infraestructura de producción, pero el equipo de soporte de PingCAP jugó un papel muy importante y nos ayudó a ajustar el clúster a la naturaleza de la carga de trabajo.

  
  

Conclusión

En esta publicación, exploramos los desafíos de usar Aurora con nuestro caso de uso de data marts en tiempo real y el viaje de migración a TiDB. También analizamos cómo Delhivery utiliza TiDB a escala.

A pesar de nuestro éxito con TiDB, reconocemos que ninguna solución es perfecta y que la efectividad puede variar según el caso de uso. En TiDB, notamos un par de áreas de mejora, incluida la falta de soporte listo para usar para vistas materializadas y administración de cuotas nativas. Sin embargo, con soluciones y ajustes adecuados, hemos logrado abordar estas limitaciones de manera efectiva.

Hasta ahora, hemos implementado TiDB en nuestro entorno de producción. Según nuestros puntos de referencia, TiDB nos permite manejar más de miles de solicitudes por segundo con una latencia de menos de 100 ms. En el futuro, continuaremos explorando más casos de uso que requieren una base de datos sólida y distribuida de manera consistente.

Referencias

https://docs.pingcap.com/tidb/stable/tidb-lightning-overview

https://reorg.github.io/pg_repack/

https://github.com/dataegret/pgcompacttable

https://cloud.google.com/spanner

https://www.yugabyte.com/yugabytedb/

https://cloud.google.com/bigquery/

https://docs.pingcap.com/tidb/dev/transaction-overview

https://proxysql.com/

Autor:

Hari Kishan (Gerente senior de ingeniería en Delhivery)

Akash Deep Verma (Director de Tecnología @ Delhivery)