¿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.

Apache Arrow DataFusion: Construyendo Query Engines Personalizados en Rust

Descripción General y Arquitectura de DataFusion

Apache Arrow DataFusion 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.

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:

#[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:

// 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.

Apache Arrow DataFusion: Construyendo Query Engines Personalizados en Rust