Apache Hudi

hudi-rs is a native Rust implementation of Apache Hudi with Python and C++ bindings. It creates, writes, and reads Hudi tables from Rust and Python in a single process, with no JVM.

Source	Downloads	Installation Command
PyPi.org		pip install hudi
Crates.io		cargo add hudi

Features

Area	Supported
Table types	Copy-on-Write (COW), Merge-on-Read (MOR)
Table versions	Read: 6, 8, 9 (Hudi 0.14 and later). Write: 8, 9 (Hudi 1.0 and later), plus a narrow append path on 6
Queries	Snapshot, time travel, incremental, read-optimized (MOR), streaming snapshot reads, column filters with pruning
Writes	Create, append, upsert, update, delete (by predicate or key), overwrite, dynamic partition overwrite
Table services	Merge-on-read compaction (experimental: on demand, inline, or in the background), rollback of failed writes, timeline archival
Metadata table	File listings, column and partition stats, record-level index, kept up to date on every write and compacted as it grows (on by default)
Base file formats	Parquet (read and write), Lance (read), HFile (metadata table)
Storage	Local file system, Amazon S3, Azure Blob Storage, Google Cloud Storage
APIs	Rust and Python (read and write); C++ and a C ABI (hudi-ffi) for file-slice reads
Query engines	Apache DataFusion (datafusion feature in Rust, included in Python); see Other Integrations

Quick start

Create a table, upsert and delete rows, then read what changed after the first commit.

Python

import tempfile

import pyarrow as pa
from hudi import HudiQueryType, HudiReadOptions, HudiTable

def docs(ids, version):
    return pa.table({
        "doc_id": ids,
        "title": [f"{d} v{version}" for d in ids],
        "version": pa.array([version] * len(ids), pa.int64()),
    })

table = HudiTable.create(f"{tempfile.mkdtemp()}/docs", record_key="doc_id", ordering_fields="version")
first = table.append(docs(["doc1", "doc2", "doc3"], version=1))

result = table.upsert(docs(["doc2", "doc4"], version=2))
print(result.num_updates, result.num_inserts)  # 1 1

table.delete("doc_id = 'doc3'")

# Commits that completed after the first one.
changed = table.read(
    HudiReadOptions()
    .with_query_type(HudiQueryType.Incremental)
    .with_start_timestamp(first.completion_time)
)
print(sorted(pa.Table.from_batches(changed).column("doc_id").to_pylist()))  # ['doc2', 'doc4']

snapshot = pa.Table.from_batches(table.read())
print(sorted(snapshot.column("doc_id").to_pylist()))  # ['doc1', 'doc2', 'doc4']

Rust

cargo add hudi
cargo add arrow@59
cargo add tokio@1 --features macros,rt-multi-thread

use std::sync::Arc;

use arrow::array::{Int64Array, RecordBatch, StringArray};
use arrow::datatypes::{DataType, Field, Schema};
use hudi::error::Result;
use hudi::table::{QueryType, ReadOptions, Table};

fn docs(ids: &[&str], version: i64) -> RecordBatch {
    let schema = Arc::new(Schema::new(vec![
        Field::new("doc_id", DataType::Utf8, false),
        Field::new("title", DataType::Utf8, false),
        Field::new("version", DataType::Int64, false),
    ]));
    let titles: Vec<String> = ids.iter().map(|d| format!("{d} v{version}")).collect();
    RecordBatch::try_new(
        schema,
        vec![
            Arc::new(StringArray::from(ids.to_vec())),
            Arc::new(StringArray::from(titles)),
            Arc::new(Int64Array::from(vec![version; ids.len()])),
        ],
    )
    .unwrap()
}

#[tokio::main]
async fn main() -> Result<()> {
    let base_uri = std::env::temp_dir().join(format!("docs_{}", std::process::id()));
    let mut table = Table::create(base_uri.to_str().unwrap())
        .with_table_name("docs")
        .with_record_key_fields(["doc_id"])
        .with_ordering_fields(["version"])
        .create()
        .await?;
    let first = table.append([docs(&["doc1", "doc2", "doc3"], 1)]).await?;

    let result = table.upsert([docs(&["doc2", "doc4"], 2)]).await?;
    println!("{} {}", result.num_updates, result.num_inserts); // 1 1

    table.delete("doc_id = 'doc3'").await?;

    // Commits that completed after the first one.
    let options = ReadOptions::new()
        .with_query_type(QueryType::Incremental)
        .with_start_timestamp(first.completion_time.unwrap());
    let changed = table.read(&options).await?;
    let snapshot = table.read(&ReadOptions::new()).await?;
    let num_rows = |batches: &[RecordBatch]| batches.iter().map(|b| b.num_rows()).sum::<usize>();
    println!("{} {}", num_rows(&changed), num_rows(&snapshot)); // 2 3
    Ok(())
}

This is the beginning of the README. Read the rest at the original. Read the full README ↗