<!-- Canonical: https://lucioduran.com/es/blog/arrow-datafusion-query-engine-rust
Language: es
Content-SHA256: 11bcc109957c7249fd4893ed1eecf44cb61ffaac925f742d58a312210a31ba09
-->

Data Engineering

# Apache Arrow DataFusion: Construyendo Query Engines Personalizados en Rust

2026-01-053 min lectura

Lucio Durán
Engineering Manager & AI Solutions Architect

También disponible en: [English](<https://lucioduran.com/blog/arrow-datafusion-query-engine-rust>)

## Descripción General y Arquitectura de DataFusion

[Apache Arrow DataFusion](<https://datafusion.apache.org/>) es un query engine, no una base de datos. Proporciona:

- Un parser y planner de SQL
- Un optimizador de queries basado en costo (con 60+ reglas de optimización)
- Un motor de ejecución paralelo usando Arrow RecordBatches
- Una API de extensión para customizar todo

Lo que NO te da: gestión de almacenamiento, transacciones, replicación, gestión de usuarios. Es una librería, no un servidor.

```rust
use datafusion::prelude::*;

#[tokio::main]
async fn main() -> datafusion::error::Result<()> {
 let ctx = SessionContext::new();

 // Registrar un archivo Parquet como tabla
 ctx.register_parquet(
 "events",
 "s3://mybucket/events/*.parquet",
 ParquetReadOptions::default(),
 ).await?;

 // SQL simplemente funciona
 let df = ctx.sql("
 SELECT date_trunc('hour', timestamp) as hour,
 count(*) as event_count,
 approx_percentile_cont(latency_ms, 0.99) as p99
 FROM events
 WHERE timestamp > '2026-02-01'
 GROUP BY 1
 ORDER BY 1
 ").await?;

 df.show().await?;
 Ok(())
}
```

Eso es una query completamente paralela con predicate pushdown a row groups de Parquet, ejecución columnar sobre arrays Arrow, y computación de percentil aproximado. En unas 15 líneas.

## Custom Table Providers: Tus Datos, Tus Reglas

Acá es donde DataFusion resulta particularmente relevante. El trait ` TableProvider ` te permite envolver cualquier fuente de datos:

```rust
#[async_trait]
impl TableProvider for MetricsTableProvider {
 fn as_any(&self) -> &dyn Any { self }

 fn schema(&self) -> SchemaRef {
 self.schema.clone()
 }

 fn table_type(&self) -> TableType {
 TableType::Base
 }

 async fn scan(
 &self,
 state: &SessionState,
 projection: Option<&Vec<usize>>,
 filters: &[Expr],
 limit: Option<usize>,
 ) -> Result<Arc<dyn ExecutionPlan>> {
 // Extraer rango temporal de los filtros para pushdown
 let time_range = extract_time_range(filters);
 let projected_schema = project_schema(&self.schema, projection);

 Ok(Arc::new(MetricsScanExec {
 schema: projected_schema,
 client: self.metrics_client.clone(),
 time_range,
 limit,
 }))
 }

 fn supports_filters_pushdown(
 &self,
 filters: &[&Expr],
 ) -> Result<Vec<TableProviderFilterPushDown>> {
 filters.iter().map(|f| {
 if is_time_filter(f) {
 Ok(TableProviderFilterPushDown::Exact)
 } else {
 Ok(TableProviderFilterPushDown::Unsupported)
 }
 }).collect()
 }
}
```

El método ` supports_filters_pushdown ` le dice al optimizador qué predicados tu provider puede manejar nativamente. Devolver ` Exact ` significa que el optimizador no va a aplicar ese filtro de nuevo después del scan, tu provider lo manejó completamente. Esto elimina filtrado redundante y puede cortar queries de métricas de ~800ms a ~200ms.

## Funciones Definidas por Usuario

DataFusion soporta UDFs escalares, de agregación, y de ventana:

```rust
// UDF escalar: distancia haversine custom
fn haversine_distance(args: &[ArrayRef]) -> Result<ArrayRef> {
 let lat1 = args[0].as_any().downcast_ref::<Float64Array>().unwrap();
 let lon1 = args[1].as_any().downcast_ref::<Float64Array>().unwrap();
 let lat2 = args[2].as_any().downcast_ref::<Float64Array>().unwrap();
 let lon2 = args[3].as_any().downcast_ref::<Float64Array>().unwrap();

 let result: Float64Array = lat1.iter().zip(lon1.iter())
 .zip(lat2.iter().zip(lon2.iter()))
 .map(|((la1, lo1), (la2, lo2))| {
 match (la1, lo1, la2, lo2) {
 (Some(la1), Some(lo1), Some(la2), Some(lo2)) => {
 Some(haversine(la1, lo1, la2, lo2))
 }
 _ => None,
 }
 })
 .collect();

 Ok(Arc::new(result))
}

ctx.register_udf(create_udf(
 "haversine",
 vec![DataType::Float64; 4],
 Arc::new(DataType::Float64),
 Volatility::Immutable,
 Arc::new(haversine_distance),
));

// Ahora usalo en SQL
let df = ctx.sql("
 SELECT city_name,
 haversine(lat, lon, -34.6037, -58.3816) as distancia_a_bsas
 FROM cities
 ORDER BY distancia_a_bsas
 LIMIT 10
").await?;
```

## Resultados de Benchmark

En una máquina de 32 cores, consultando 50 GB de datos Parquet (1.2 mil millones de filas):

```
Query: SELECT date_trunc('hour', ts), count(*), avg(value)
 FROM events WHERE ts > '2026-02-01' GROUP BY 1

DuckDB (in-process): 1,8s
DataFusion (default): 2,4s
DataFusion (tuneado): 1,1s
Spark (local[32]): 14,2s
```

El DataFusion tuneado fue en realidad más rápido que DuckDB en este workload específico porque el reader Parquet custom solapaba I/O con computación, mientras DuckDB bloqueaba en lecturas, DataFusion estaba procesando el batch anterior.

Para analytics sobre datos heterogéneos sin un servidor de base de datos administrado, DataFusion es una de las opciones más sólidas disponibles. El ecosistema Rust alrededor de Arrow está madurando rápido, y la comunidad es muy activa. Las APIs de extensión son poderosas pero no triviales, conocimiento sólido de Rust es un prerrequisito.

## Preguntas frecuentes

### ¿Qué es Apache Arrow DataFusion?

DataFusion es un query engine extensible escrito en Rust, construido sobre el formato de memoria columnar de Apache Arrow. Provee una API de SQL y DataFrame, un optimizador de queries basado en costo, y un motor de ejecución paralelo que es posible embeber en tus propias aplicaciones o extender con fuentes de datos custom, funciones y estrategias de ejecución.

### ¿Cuándo debería usar DataFusion en vez de una base de datos existente?

DataFusion sobresale cuando se necesita analytics embebidos sin un servidor de base de datos separado, fuentes de datos custom que no encajan en bases de datos tradicionales, o cuando se desea construir query engines específicos de dominio. Si se necesita transacciones, concurrencia multi-usuario, o garantías ACID completas, una base de datos tradicional es mejor.

### ¿Cuál es la diferencia entre ejecución push-based y pull-based?

En ejecución pull-based (Volcano), cada operador tira batches de sus hijos llamando next(). En ejecución push-based, los datos fluyen downstream sin que los operadores necesiten solicitarlos. DataFusion usa un modelo de streaming pull-based con RecordBatch como unidad de trabajo, que soporta backpressure naturalmente y se integra bien con el modelo async de Rust.

## Tips

1. **Tip:** El trait ` TableProvider ` de DataFusion es engañosamente poderoso. Puede envolver literalmente cualquier fuente de datos, archivos Parquet en S3, APIs REST, hasta streams gRPC, detrás de él, y el optimizador de queries simplemente funciona. Un provider custom para un store interno de métricas se puede armar en unas 200 líneas de Rust.
2. **Tip:** Si tus queries filtran por una columna específica el 90% del tiempo, implementá ` supports_filters_pushdown ` en tu ` TableProvider `. El optimizador va a pushear predicados hasta tu scan, y es posible saltear row groups enteros. Solo esto puede cortar la latencia de queries 4x.
3. **Tip:** No escribas funciones de agregación custom desde cero. Utilizar el trait ` Accumulator ` de DataFusion, maneja particionamiento y merge entre particiones automáticamente. Armar un HyperLogLog paralelo custom desde cero es innecesario cuando el trait maneja todas las partes difíciles automáticamente.

- arrow
- datafusion
- rust
- query-engine
- columnar
- data-engineering
- analytics
- performance

[Volver al Blog](<https://lucioduran.com/es/blog>)