Sincronización de alto rendimiento con el formato de almacenamiento columnar Arrow - DataWorks

La integración de datos de DataWorks admite la sincronización de datos de alto rendimiento entre distintas fuentes mediante el formato de almacenamiento columnar Apache Arrow. Esta solución utiliza transferencia directa en memoria y tecnología de copia cero para transformar la transferencia de datos de un modelo basado en filas a uno basado en columnas, lo que mejora el throughput de sincronización para grandes volúmenes de datos. En escenarios de prueba, el rendimiento puede aumentar casi 10 veces.

Casos de uso

Al procesar grandes cantidades de datos, las soluciones de sincronización basadas en almacenamiento por filas suelen enfrentar cuellos de botella de rendimiento. Estos se deben a la serialización, deserialización y conversión de tipos frecuentes, especialmente durante migraciones de almacenes de datos (como de Hive a MaxCompute) o al construir una arquitectura de data lakehouse. Los tiempos prolongados de sincronización para terabytes de datos pueden afectar el lanzamiento de servicios y la eficiencia del análisis de datos. Esta solución resuelve la ineficiencia en la transferencia de datos entre fuentes de almacenamiento columnar a gran escala y reduce el tiempo de sincronización de horas a minutos.

Escenario 1: Migración de big data

Problema: Migrar cientos de terabytes de datos de Hive a MaxCompute consume mucho tiempo y retrasa el lanzamiento de servicios. Solución: Habilite la sincronización Arrow para transferir directamente el almacenamiento columnar y evitar la conversión de formatos. Resultado: El tiempo de migración se reduce de horas a minutos, lo que aumenta la eficiencia en más de 10 veces.

Escenario 2: Integración de fuentes de datos diversas y arquitectura de data lakehouse

Los datos fluyen de manera libre y eficiente entre el data lake (como el ecosistema de Hive) y el almacén de datos (Hologres/MaxCompute). El valor principal consiste en usar una sola copia de los datos para múltiples propósitos y habilitar la sinergia entre el lake y el warehouse.

Arquitectura de la solución

La integración de datos de DataWorks incorpora el estándar columnar en memoria Apache Arrow. Esto refactoriza el método de transferencia de datos entre el Reader y el Writer, y cambia el proceso de un tratamiento basado en filas a una transferencia directa basada en columnas.

Arquitectura tradicional de sincronización con almacenamiento por filas

En la arquitectura tradicional, los datos se procesan fila por fila. Aunque tanto el origen como el destino utilicen un formato de almacenamiento columnar, como Parquet u ORC, los datos se convierten en objetos Record genéricos basados en filas dentro del motor de sincronización. Este formato está diseñado para el almacenamiento de una sola fila. Cada objeto Record define varias Columnas, y cada Columna contiene el Valor correspondiente a esa columna en la fila actual. Por ejemplo, considere la sincronización de datos de almacenamiento columnar desde MaxCompute (ODPS) hacia MaxCompute (ODPS):

Este proceso incluye los siguientes pasos y genera la siguiente sobrecarga de rendimiento:

Decodificación del Reader: El Reader lee datos de almacenamiento columnar desde el origen, como MaxCompute. Luego decodifica los datos y los convierte fila por fila en objetos Record internos del framework.
Transferencia del framework: El framework transfiere una gran cantidad de objetos Record en la cola de memoria.
Codificación del Writer: El Writer recupera los objetos Record del framework, los codifica al formato de almacenamiento columnar que requiere el destino y luego escribe los datos.

Este proceso implica múltiples conversiones de formato de datos, serialización y deserialización, además de creación frecuente de objetos. Como consecuencia, se produce un alto consumo de recursos de CPU y memoria, recolección de basura (GC) frecuente y un throughput general limitado.

Arquitectura de sincronización con almacenamiento columnar Arrow

La nueva arquitectura basada en Arrow implementa un flujo de datos columnar de extremo a extremo. Los datos mantienen su formato de almacenamiento columnar a lo largo de todo el pipeline de sincronización, lo que evita la sobrecarga de la conversión basada en filas.

Al introducir la estructura de datos ArrowTabularRecord, el proceso de sincronización se optimiza de la siguiente manera:

Lectura directa por el Reader: El Reader usa la API Tunnel Arrow de MaxCompute para leer directamente datos de almacenamiento columnar desde el origen. Después encapsula los datos en objetos ArrowTabularRecord y los entrega al framework en lotes.
Transferencia del framework: El framework pasa directamente los objetos ArrowTabularRecord, que contienen datos binarios columnares.
Escritura directa por el Writer: El Writer usa la API Tunnel Arrow para recuperar datos binarios en formato Arrow directamente desde el ArrowTabularRecord y utiliza tecnología de copia cero para transferirlos al destino, como un Tunnel Server, sin volver a serializarlos.

Esta arquitectura proporciona soporte nativo para Arrow y elimina el paso intermedio de conversión basada en filas. De este modo se logra una "sincronización en cortocircuito" de extremo a extremo para almacenes columnares, lo que mejora significativamente el throughput y reduce la latencia.

Pasos de implementación

Para habilitar la sincronización Arrow de alto rendimiento, agregue parámetros específicos a una tarea de integración de datos de DataWorks. Actualmente, esta función es compatible con tareas de sincronización offline de base de datos completa y de tabla única donde el origen y el destino son MaxCompute, Hologres o Hive/OSS/Hadoop Distributed File System (HDFS) (Parquet/ORC).

Método 1: Sincronización offline de base de datos completa

Utilice la función full-database offline synchronization task de integración de datos. Por ejemplo, en una sincronización offline de Hive a MaxCompute, el sistema detecta automáticamente los tipos de campo de las tablas de origen y destino, y habilita la sincronización Arrow de alto rendimiento. No se requiere configuración manual. Consulte este ajuste en Advanced Configuration > Advanced Parameters > Runtime Config, ubicado en la esquina superior derecha del editor de tareas.

En Runtime Config, localice el parámetro Enable Arrow short-circuit transfer. Las opciones disponibles son Auto (predeterminado), Force enable y Force disable.

Método 2: Sincronización offline de tabla única

Al configurar una tarea de sincronización offline de tabla única, habilite manualmente la función en code editor. Para ello, agregue "useArrow": true a la configuración de parameter tanto del Reader como del Writer.

Requisito previo: Para habilitar esta función, las definiciones de tipos de columna de las tablas de origen y destino deben ser idénticas. Esto se debe a que este modo omite el paso de conversión de tipos y transfiere los datos directamente en memoria.

A continuación se muestra un ejemplo de configuración para sincronizar datos desde Hive (Reader) hacia MaxCompute (Writer):

{
  "type": "job",
  "steps": [
    {
      "stepType": "hive",
      "parameter": {
        "useArrow": true,  
        "datasource": "my_datasource",
        "column": [
          "col1",
          "col2"
        ],
        "readMode": "hdfs",
        "table": "table"
      },
      "name": "Reader",
      "category": "reader"
    },
    {
      "stepType": "odps",
      "parameter": {
        "useArrow": true,  
        "truncate": false,
        "datasource": "odps_test",
        "column": [
          "col1",
          "col2"
        ],
        "table": "table"
      },
      "name": "Writer",
      "category": "writer"
    }
  ],
  "setting": {
    "speed": {
      "concurrent": 3
    }
  }
}

Validación de rendimiento

Las siguientes secciones muestran la mejora de rendimiento para varias combinaciones de fuentes de datos. Las condiciones de prueba para la validación de rendimiento son las siguientes:

Condiciones de prueba de estrés: Especificaciones del servidor: 64 núcleos, 256 GB de memoria. Tarjeta de red: 25 Gbit/s.
Volumen de datos de prueba: 42,045,700 registros.

Fuente de datos	Función compatible	Mejora en el rendimiento de replicación
MaxCompute	Lectura directa de datos de almacenamiento columnar mediante la API Tunnel Arrow	El rendimiento de replicación mejoró un 200%
Hologres	Admite exportación en formato Arrow para evitar cuellos de botella basados en filas de Java Database Connectivity (JDBC)	El rendimiento de replicación mejoró un 95%
Hive, OSS, HDFS y otros sistemas de archivos distribuidos	Lectura directa de datos subyacentes en formato Arrow desde Parquet/ORC	El rendimiento de sincronización PARQUET mejoró 5.55 veces. El rendimiento de sincronización ORC mejoró 9.85 veces.

Escenario 1: Sincronización en cortocircuito de almacenamiento columnar MaxCompute (Arrow → Arrow)

Concurrencia	Almacenamiento por filas tradicional	Almacenamiento columnar Arrow	Mejora de rendimiento
1	67.8 MB/s 3740 R/s	212.6 MB/s 11462 R/s	+206.5%
3	185.6 MB/s 10226 R/s	569.9 MB/s 30728 R/s	+200.5%
8	462.1 MB/s 25467 R/s	1321.0 MB/s 71143 R/s	+197.4%

Escenario 2: Sincronización Hologres → MaxCompute

Concurrencia

Sincronización tradicional

Sincronización Arrow

Mejora de rendimiento

439.1 MB/s

216480 R/s

906.1 MB/s

404270 R/s

+87%

773.3 MB/s

381300 R/s

1669.1 MB/s

745654 R/s

+95%

Escenario 3: Sincronización Parquet/ORC → MaxCompute

Formato de almacenamiento columnar

Sincronización tradicional

Sincronización Arrow

Mejora de rendimiento

Parquet

26.1 MB/s

35631 R/s

1198.1 MB/s

233587 R/s

5.55 veces

ORC

21.4 MB/s

27661 R/s

3256.3 MB/s

300326 R/s

9.85 veces

Parquet y ORC son formatos de almacenamiento columnar utilizados en sistemas de archivos distribuidos como HDFS y OSS.