Rust is gaining traction in data engineering for the same reasons it succeeds in systems programming: predictable performance, zero-copy data access, and memory safety without a GC. This matters in data workloads where latency spikes, memory pressure, and CPU throughput directly affect pipeline cost.
Why Rust for Data Engineering
Traditional data engineering stacks (Spark, Pandas, Flink) are built on JVM or Python — both with GC pauses and high per-record overhead. Rust offers:
- Zero-copy reads: Apache Arrow’s columnar format + Rust’s ownership means no unnecessary data copies between pipeline stages
- Predictable latency: No GC pauses — important for streaming pipelines with SLAs
- Memory efficiency: Rust programs typically use 2–5x less memory than equivalent JVM programs
- WASM and edge: Rust compiles to WebAssembly, enabling data processing at the edge or in browsers
- FFI with Python: PyO3 lets you write performance-critical pipeline stages in Rust and call them from Python
The ecosystem has matured significantly: Apache Arrow, DuckDB, Polars, DataFusion, and Lance all have first-class Rust bindings.
Apache Arrow: Columnar In-Memory Format
Apache Arrow is the de facto standard for in-memory columnar data. It eliminates serialization overhead when passing data between systems and enables SIMD-optimized operations.
# Cargo.toml
[dependencies]
arrow = "53.0"
arrow-array = "53.0"
arrow-schema = "53.0"
arrow-csv = "53.0"
Creating RecordBatches
use arrow_array::{Int32Array, StringArray, Float64Array, RecordBatch, ArrayRef};
use arrow_schema::{Schema, Field, DataType};
use std::sync::Arc;
fn create_users_batch() -> RecordBatch {
let schema = Arc::new(Schema::new(vec![
Field::new("id", DataType::Int32, false),
Field::new("name", DataType::Utf8, false),
Field::new("score", DataType::Float64, true), // nullable
]));
let ids = Int32Array::from(vec![1, 2, 3, 4, 5]);
let names = StringArray::from(vec!["alice", "bob", "carol", "dave", "eve"]);
let scores = Float64Array::from(vec![
Some(92.5), Some(87.0), None, Some(95.0), Some(78.5)
]);
RecordBatch::try_new(
schema,
vec![
Arc::new(ids) as ArrayRef,
Arc::new(names) as ArrayRef,
Arc::new(scores) as ArrayRef,
],
).unwrap()
}
Filtering and Projecting
use arrow::compute::{filter, gt_scalar, cast};
use arrow_array::BooleanArray;
fn filter_high_scorers(batch: &RecordBatch, threshold: f64) -> RecordBatch {
// Get the scores column
let scores = batch.column_by_name("score").unwrap();
let scores_f64 = scores.as_any().downcast_ref::<Float64Array>().unwrap();
// Build boolean mask: score > threshold
let mask = gt_scalar(scores_f64, threshold).unwrap();
// Apply filter to all columns
let filtered_cols: Vec<ArrayRef> = batch.columns().iter()
.map(|col| filter(col.as_ref(), &mask).unwrap())
.collect();
RecordBatch::try_new(batch.schema(), filtered_cols).unwrap()
}
fn main() {
let batch = create_users_batch();
let high_scorers = filter_high_scorers(&batch, 85.0);
println!("High scorers: {} rows", high_scorers.num_rows()); // 3
}
Reading and Writing CSV
use arrow::csv::{ReaderBuilder, WriterBuilder};
use std::fs::File;
use std::io::BufReader;
fn read_csv(path: &str) -> Vec<RecordBatch> {
let file = File::open(path).unwrap();
let reader = ReaderBuilder::new(Arc::new(Schema::empty()))
.has_header(true)
.infer_schema(Some(100)) // infer from first 100 rows
.build(BufReader::new(file))
.unwrap();
reader.collect::<Result<Vec<_>, _>>().unwrap()
}
fn write_csv(path: &str, batches: &[RecordBatch]) {
let file = File::create(path).unwrap();
let mut writer = WriterBuilder::new().build(file);
for batch in batches {
writer.write(batch).unwrap();
}
writer.finish().unwrap();
}
Reading Parquet (via parquet crate)
parquet = "53.0"
use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
use parquet::arrow::ArrowWriter;
use std::fs::File;
fn read_parquet(path: &str) -> Vec<RecordBatch> {
let file = File::open(path).unwrap();
let builder = ParquetRecordBatchReaderBuilder::try_new(file).unwrap();
let reader = builder.with_batch_size(8192).build().unwrap();
reader.collect::<Result<Vec<_>, _>>().unwrap()
}
fn write_parquet(path: &str, batches: &[RecordBatch]) {
let file = File::create(path).unwrap();
let schema = batches[0].schema();
let mut writer = ArrowWriter::try_new(file, schema, None).unwrap();
for batch in batches {
writer.write(batch).unwrap();
}
writer.close().unwrap();
}
DuckDB: In-Process Analytics
DuckDB is a fast in-process analytical database that works natively with Arrow. It’s perfect for ad-hoc analytics, ETL transformations, and replacing Pandas for large datasets.
duckdb = { version = "1.0", features = ["bundled"] }
Basic Queries
use duckdb::{Connection, Result, params};
fn analytics_example() -> Result<()> {
let conn = Connection::open_in_memory()?;
// Create and populate table
conn.execute_batch("
CREATE TABLE orders AS SELECT * FROM read_csv_auto('orders.csv');
")?;
// Analytical query — aggregate by month
let mut stmt = conn.prepare("
SELECT
strftime('%Y-%m', order_date) AS month,
COUNT(*) AS order_count,
SUM(amount) AS total_amount,
AVG(amount) AS avg_amount
FROM orders
WHERE status = ?
GROUP BY month
ORDER BY month
")?;
let rows = stmt.query_map(params!["completed"], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, i64>(1)?,
row.get::<_, f64>(2)?,
row.get::<_, f64>(3)?,
))
})?;
for row in rows {
let (month, count, total, avg) = row?;
println!("{}: {} orders, total={:.2}, avg={:.2}", month, count, total, avg);
}
Ok(())
}
DuckDB + Apache Arrow (Zero-Copy)
DuckDB can query Arrow RecordBatches directly with no serialization:
use duckdb::arrow::record_batch::RecordBatch;
use duckdb::{Connection, Result};
fn query_arrow_data(batches: Vec<RecordBatch>) -> Result<()> {
let conn = Connection::open_in_memory()?;
// Register Arrow data as a DuckDB table
conn.register_arrow("my_data", batches)?;
// Query it with SQL
let mut stmt = conn.prepare("
SELECT name, COUNT(*) as cnt
FROM my_data
WHERE score > 80
GROUP BY name
ORDER BY cnt DESC
LIMIT 10
")?;
let result: Vec<RecordBatch> = stmt.query_arrow([])?.collect();
println!("Result batches: {}", result.len());
Ok(())
}
File Format Support
DuckDB’s secret weapon is native support for Parquet, CSV, JSON, and even remote S3 files:
fn query_files(conn: &Connection) -> Result<()> {
// Query Parquet directly
conn.execute_batch("
SELECT * FROM 'data/*.parquet'
WHERE date >= '2026-01-01'
LIMIT 1000;
")?;
// Query S3 (requires httpfs extension)
conn.execute_batch("
INSTALL httpfs;
LOAD httpfs;
SELECT COUNT(*) FROM 's3://my-bucket/data/events/*.parquet';
")?;
Ok(())
}
Polars: DataFrame Operations
Polars is a DataFrame library built on Apache Arrow with a lazy evaluation engine. It’s significantly faster than Pandas for most workloads.
polars = { version = "0.42", features = ["lazy", "csv", "parquet", "json"] }
Eager Operations
use polars::prelude::*;
fn polars_basics() -> PolarsResult<()> {
// Create DataFrame
let df = df!(
"name" => ["Alice", "Bob", "Carol", "Dave", "Eve"],
"dept" => ["Eng", "Eng", "Sales", "Sales", "Eng"],
"salary" => [95000i32, 87000, 72000, 68000, 91000],
"years" => [5i32, 3, 7, 2, 4],
)?;
println!("{}", df);
// Filter
let senior = df.filter(&df.column("years")?.gt(3)?)?;
println!("Senior employees:\n{}", senior);
// Sort
let sorted = df.sort(["salary"], SortMultipleOptions::default().with_order_descending(true))?;
println!("By salary:\n{}", sorted);
// Select columns
let subset = df.select(["name", "salary"])?;
println!("Subset:\n{}", subset);
Ok(())
}
Lazy Evaluation (the Fast Path)
fn polars_lazy() -> PolarsResult<()> {
// Build a lazy query plan — nothing executes yet
let q = LazyFrame::scan_csv("employees.csv", ScanArgsAnonymous::default())?
.filter(col("dept").eq(lit("Engineering")))
.with_column(
(col("salary") * lit(1.1)).alias("adjusted_salary")
)
.group_by([col("team")])
.agg([
col("adjusted_salary").mean().alias("avg_salary"),
col("name").count().alias("headcount"),
])
.sort("avg_salary", SortOptions::default().with_order_descending(true))
.limit(10);
// Execute the optimized plan
let result = q.collect()?;
println!("{}", result);
Ok(())
}
Reading Multiple Files
fn polars_multi_file() -> PolarsResult<()> {
// Read all Parquet files in a directory
let df = LazyFrame::scan_parquet("data/events/*.parquet", Default::default())?
.filter(col("event_type").eq(lit("purchase")))
.select([col("user_id"), col("amount"), col("timestamp")])
.collect()?;
println!("Loaded {} events", df.height());
Ok(())
}
Building an Async ETL Pipeline
A production-grade ETL pipeline using Tokio channels and Arrow:
tokio = { version = "1", features = ["full"] }
arrow = "53.0"
use arrow_array::RecordBatch;
use tokio::sync::mpsc;
use std::sync::Arc;
// Each stage processes batches and forwards them
type BatchSender = mpsc::Sender<RecordBatch>;
type BatchReceiver = mpsc::Receiver<RecordBatch>;
async fn extract_stage(sender: BatchSender, source: &str) {
// Simulate reading from a data source in chunks
for chunk_id in 0..10 {
let batch = read_chunk(source, chunk_id);
if sender.send(batch).await.is_err() {
break; // Downstream closed
}
}
}
async fn transform_stage(mut receiver: BatchReceiver, sender: BatchSender) {
while let Some(batch) = receiver.recv().await {
// Apply transformations
let transformed = normalize_batch(batch);
if sender.send(transformed).await.is_err() {
break;
}
}
}
async fn load_stage(mut receiver: BatchReceiver, destination: &str) {
let mut writer = open_parquet_writer(destination);
while let Some(batch) = receiver.recv().await {
writer.write(&batch).unwrap();
}
writer.close().unwrap();
}
#[tokio::main]
async fn main() {
let (extract_tx, extract_rx) = mpsc::channel::<RecordBatch>(32);
let (transform_tx, transform_rx) = mpsc::channel::<RecordBatch>(32);
let extract = tokio::spawn(extract_stage(extract_tx, "s3://source/data/"));
let transform = tokio::spawn(transform_stage(extract_rx, transform_tx));
let load = tokio::spawn(load_stage(transform_rx, "output/result.parquet"));
// Wait for pipeline to complete
let (e, t, l) = tokio::join!(extract, transform, load);
e.unwrap(); t.unwrap(); l.unwrap();
println!("ETL complete");
}
// Placeholder implementations
fn read_chunk(_source: &str, _id: u32) -> RecordBatch { todo!() }
fn normalize_batch(batch: RecordBatch) -> RecordBatch { batch }
fn open_parquet_writer(_path: &str) -> impl std::io::Write { std::io::sink() }
Parallel Batch Processing with Rayon
For CPU-bound transformations, process batches in parallel:
rayon = "1.10"
use rayon::prelude::*;
use arrow_array::RecordBatch;
fn parallel_transform(batches: Vec<RecordBatch>) -> Vec<RecordBatch> {
batches.into_par_iter()
.map(|batch| {
// Each batch processed on its own thread
apply_transformations(batch)
})
.collect()
}
fn apply_transformations(batch: RecordBatch) -> RecordBatch {
// normalize, filter, enrich...
batch
}
DataFusion: SQL Query Engine
Apache DataFusion is a full SQL query engine built on Arrow and Tokio, written in Rust. Use it to build custom query engines or add SQL interfaces to your data platform:
datafusion = "42.0"
use datafusion::prelude::*;
#[tokio::main]
async fn main() -> datafusion::error::Result<()> {
let ctx = SessionContext::new();
// Register a Parquet file as a table
ctx.register_parquet("employees", "data/employees.parquet", Default::default()).await?;
// Execute SQL
let df = ctx.sql("
SELECT dept, AVG(salary) as avg_salary, COUNT(*) as headcount
FROM employees
WHERE years_experience > 2
GROUP BY dept
ORDER BY avg_salary DESC
").await?;
df.show().await?;
Ok(())
}
Performance Patterns
Zero-Copy Record Access
use arrow_array::{RecordBatch, Int64Array};
use arrow::datatypes::DataType;
fn sum_column_zero_copy(batch: &RecordBatch, col_name: &str) -> i64 {
let col = batch.column_by_name(col_name).unwrap();
let array = col.as_any().downcast_ref::<Int64Array>().unwrap();
// Direct access to Arrow's buffer — no allocation
array.values().iter().sum()
}
Chunked Processing to Control Memory
fn process_large_file(path: &str, batch_size: usize) {
let file = std::fs::File::open(path).unwrap();
let reader = ParquetRecordBatchReaderBuilder::try_new(file)
.unwrap()
.with_batch_size(batch_size) // Process 8192 rows at a time
.build()
.unwrap();
let mut total_rows = 0;
for batch in reader {
let batch = batch.unwrap();
total_rows += batch.num_rows();
// Process batch — only batch_size rows in memory at a time
process_batch(&batch);
}
println!("Processed {} rows total", total_rows);
}
fn process_batch(_batch: &RecordBatch) { /* ... */ }
Summary
| Tool | Use case |
|---|---|
| Apache Arrow | In-memory columnar format, zero-copy inter-op |
| arrow-csv / parquet | Read/write files in Arrow format |
| DuckDB | SQL analytics on Arrow data, S3/Parquet queries |
| Polars | DataFrame operations, lazy evaluation, multi-file reads |
| DataFusion | Build custom SQL query engines |
| Rayon | Parallel batch processing (CPU-bound) |
| Tokio channels | Async ETL pipelines (I/O-bound stages) |
Rust’s strength in data engineering is the combination of zero-copy memory access through Arrow, predictable performance without GC pauses, and the ability to safely parallelize across all CPU cores. For data volumes where Python slows down and JVM overhead matters, Rust is increasingly the right choice.
Resources
- Apache Arrow Rust docs
- DuckDB Rust bindings
- Polars documentation
- DataFusion
- Rust data engineering community
Comments