turboframe 0.2.1

Fast parallel DataFrame library built on PyArrow for Delta Lake and Fabric

TurboFrame

A fast, parallel DataFrame library built on PyArrow. Designed for Delta Lake reads and parallel GroupBy operations on Microsoft Fabric notebooks — but works everywhere Python runs.

Why TurboFrame?

Feature	TurboFrame	Pandas	PySpark
Parallel GroupBy	Yes (multi-core hash partition)	No (single thread)	Yes (cluster)
Delta Lake read	Zero-copy via Rust	Slow (via pandas)	Fast (JVM)
Predicate pushdown	Yes (skips row groups)	No	Yes
CSV read	C++ native (PyArrow)	Python-based	JVM overhead
Memory model	Columnar (Arrow)	Row-ish	Distributed
Startup time	Instant	Instant	5-15 seconds
Best for	1GB - 50GB on single node	< 1GB	> 50GB distributed

---

Installation

# Core (parquet, csv, json support included)
pip install turboframe

# With Delta Lake support (recommended for Fabric)
pip install turboframe[delta]

# With Excel support
pip install turboframe[excel]

# With SQL support
pip install turboframe[sql]

# Everything
pip install turboframe[all]

---

Quick Start

from turboframe import TurboFrame

# Create from dict
tf = TurboFrame({"region": ["East", "West", "East"], "sales": [100, 200, 150]})

# Filter + GroupBy + Sort
result = (
    tf.filter("sales > 100")
      .groupby("region")
      .agg({"sales": "sum"})
      .sort("sales_sum", ascending=False)
)

result.show()
# Output:
# region  sales_sum
#   West        200
#   East        150

---

Reading Data

TurboFrame reads from any data source. Every reader prints timing info automatically.

Delta Table (Fastest for Fabric)

# Basic read
tf = TurboFrame.read_delta("/lakehouse/default/Tables/sales")
# [TurboFrame] Read 5,000,000 rows x 12 cols in 1.23s from delta

# Column pruning — only loads 3 columns instead of 50 (huge speedup)
tf = TurboFrame.read_delta(
    "/lakehouse/default/Tables/sales",
    columns=["region", "amount", "status"]
)

# Predicate pushdown — skips Parquet row groups that don't match
# The data never even gets loaded into memory
tf = TurboFrame.read_delta(
    "/lakehouse/default/Tables/sales",
    columns=["region", "amount", "status"],
    filters=[("amount", ">", 1000), ("status", "=", "active")]
)

# Filter with IN operator
tf = TurboFrame.read_delta(
    "/Tables/sales",
    filters=[("region", "in", ["East", "West", "North"])]
)

# OR conditions (list of lists)
tf = TurboFrame.read_delta(
    "/Tables/sales",
    filters=[
        [("region", "=", "East"), ("amount", ">", 500)],    # East AND amount>500
        [("region", "=", "West"), ("amount", ">", 1000)],   # OR West AND amount>1000
    ]
)

# Row limit — prevent out-of-memory on huge tables
tf = TurboFrame.read_delta(
    "/Tables/sales",
    columns=["region", "amount"],
    filters=[("amount", ">", 1000)],
    row_limit=500_000
)

# Time travel — read a specific Delta version
tf = TurboFrame.read_delta("/Tables/sales", version=5)

# Cloud paths
tf = TurboFrame.read_delta("abfss://[e-mail]/Tables/sales")
tf = TurboFrame.read_delta("s3://bucket/tables/sales")

Filter Operators for read_delta

Operator	Example
= or ==	("region", "=", "East")
!=	("status", "!=", "cancelled")
>	("amount", ">", 1000)
>=	("date", ">=", "2024-01-01")
<	("quantity", "<", 10)
<=	("score", "<=", 100)
in	("region", "in", ["East", "West"])
not in	("status", "not in", ["cancelled", "draft"])

Huge Tables (Batch Processing)

For tables too large to fit in memory, process in batches:

# Process a 100M row table in 200K row chunks — never OOM
results = TurboFrame.read_delta_batched(
    "/Tables/huge_table",
    batch_fn=lambda batch: batch.groupby("region").agg({"amount": "sum"}),
    columns=["region", "amount"],
    filters=[("status", "=", "active")],
    batch_size=200_000
)
# [TurboFrame] Processed 100,000,000 rows in 500 batches in 45.67s

# Merge partial results
final = results.groupby("region").agg({"amount_sum": "sum"})
final.show()

CSV / TSV

tf = TurboFrame.read_csv("data/sales.csv")
tf = TurboFrame.read_csv("data/export.tsv", delimiter="\t")
tf = TurboFrame.read_csv("data/big.csv", columns=["id", "amount"])  # column pruning

Parquet

tf = TurboFrame.read_parquet("data/sales.parquet")
tf = TurboFrame.read_parquet("data/partitioned/", columns=["region", "amount"])

JSON

# Newline-delimited JSON (JSONL) — C++ fast reader
tf = TurboFrame.read_json("data/events.jsonl")

# Standard JSON array: [{"a": 1}, {"a": 2}]
tf = TurboFrame.read_json_array("data/records.json")

Excel

tf = TurboFrame.read_excel("report.xlsx")
tf = TurboFrame.read_excel("report.xlsx", sheet_name="Q4 Data")
tf = TurboFrame.read_excel("report.xlsx", columns=["region", "revenue"])

SQL Database

# Via pyodbc / SQLAlchemy (uses pandas internally)
import pyodbc
conn = pyodbc.connect("DRIVER={ODBC Driver 18 for SQL Server};SERVER=...")
tf = TurboFrame.read_sql("SELECT * FROM sales WHERE year = 2024", conn)

# Via ADBC — fastest (reads directly into Arrow, no pandas)
import adbc_driver_postgresql.dbapi
conn = adbc_driver_postgresql.dbapi.connect("postgresql://user:pass@host/db")
tf = TurboFrame.read_sql_arrow("SELECT * FROM sales", conn)

From DataFrames

# Pandas
import pandas as pd
tf = TurboFrame(pd.DataFrame({"a": [1, 2, 3]}))
tf = TurboFrame.from_pandas(my_pandas_df)

# Spark (on Fabric notebooks)
spark_df = spark.sql("SELECT * FROM lakehouse.sales")
tf = TurboFrame(spark_df)
tf = TurboFrame.from_spark(spark_df)

# Polars (zero-copy)
import polars as pl
tf = TurboFrame(pl.read_csv("data.csv"))
tf = TurboFrame.from_polars(my_polars_df)

# Dict
tf = TurboFrame({"region": ["E", "W"], "sales": [100, 200]})
tf = TurboFrame.from_dict({"region": ["E", "W"], "sales": [100, 200]})

# List of dicts
tf = TurboFrame([{"name": "Alice", "score": 95}, {"name": "Bob", "score": 87}])
tf = TurboFrame.from_records([{"name": "Alice", "score": 95}])

---

Operations

Filter

tf.filter("amount > 1000")
tf.filter("region == 'East'")
tf.filter("status != 'cancelled'")

# Chain multiple filters (AND logic)
result = (
    tf.filter("amount > 1000")
      .filter("region == 'East'")
      .filter("quantity >= 10")
)

Select / Drop / Rename Columns

tf.select(["region", "amount", "date"])
tf.drop("internal_id")
tf.drop(["col1", "col2", "col3"])
tf.rename({"amt": "amount", "qty": "quantity"})

Add / Replace Column

tf.with_column("tax", [10, 20, 30])          # from list
tf.with_column("flag", "active")              # broadcast scalar
tf.with_column("doubled", some_arrow_array)   # from Arrow array

Sort

tf.sort("amount")                                  # ascending
tf.sort("amount", ascending=False)                 # descending
tf.sort(["region", "amount"], ascending=[True, False])  # multi-column

GroupBy (Parallel)

This is the core feature. GroupBy uses hash partitioning across all CPU cores:

1. Data is split into N partitions (one per CPU core) by hashing the group key
2. Each core aggregates its partition independently
3. Results are concatenated (no merge needed — hash guarantees no key overlap)

# Single aggregation
result = tf.groupby("region").agg({"sales": "sum"})

# Multiple aggregations
result = tf.groupby("region").agg({
    "sales": "sum",
    "quantity": "mean",
    "order_id": "count"
})

# Multiple group keys
result = tf.groupby(["region", "year"]).agg({
    "sales": "sum",
    "profit": "mean"
})

# Shorthand
tf.groupby("region").count()
tf.groupby("region").sum(["sales", "quantity"])

Supported Aggregations

Aggregation	Description
sum	Sum of values
mean	Average
min	Minimum
max	Maximum
count	Count non-null values
std	Standard deviation
var	Variance
median	Approximate median
nunique	Count distinct values

Join

sales = TurboFrame.read_delta("/Tables/sales")
regions = TurboFrame.read_delta("/Tables/regions")

sales.join(regions, on="region_id", how="inner")
sales.join(regions, on="region_id", how="left")
sales.join(regions, on="region_id", how="right")
sales.join(regions, on="region_id", how="outer")

Head / Tail / Sample

tf.head(10)
tf.tail(10)
tf.sample(100, seed=42)

Stats & Inspection

tf.shape              # (1000000, 12)
tf.columns            # ["region", "amount", ...]
tf.dtypes             # {"region": string, "amount": float64, ...}
tf.describe()         # {"amount": {"mean": 500, "min": 1, "max": 999, ...}}
tf.unique("region")   # ["East", "West", "North"]
tf.value_counts("region")
tf.count_nulls()      # {"region": 0, "amount": 5}
tf.is_empty()         # False
tf.show(20)           # pretty-print first 20 rows

---

Writing Data

# To Delta table
tf.to_delta("/Tables/output", mode="overwrite")
tf.to_delta("/Tables/output", mode="append")

# To files
tf.to_parquet("output.parquet")
tf.to_csv("output.csv")
tf.to_json("output.json")

# To other frameworks
pdf = tf.to_pandas()
pldf = tf.to_polars()
sdf = tf.to_spark(spark)
arrow = tf.to_arrow()
d = tf.to_dict()

---

Complete Pipeline Examples

Example 1: Sales Analysis on Fabric

from turboframe import TurboFrame

# Read with column pruning + predicate pushdown
tf = TurboFrame.read_delta(
    "/lakehouse/default/Tables/transactions",
    columns=["region", "category", "amount", "date", "status"],
    filters=[("status", "=", "completed"), ("amount", ">", 0)]
)

# Parallel groupby
summary = (
    tf.groupby(["region", "category"])
      .agg({"amount": "sum", "date": "count"})
      .sort("amount_sum", ascending=False)
)

summary.show(20)

# Write results back to Delta
summary.to_delta("/lakehouse/default/Tables/sales_summary")

Example 2: Spark Pre-filter + TurboFrame Analytics

# Best pattern for huge tables on Fabric:
# Spark filters across the cluster, TurboFrame does fast local analytics

spark_df = spark.sql("""
    SELECT region, category, amount, quantity
    FROM lakehouse.transactions
    WHERE year = 2024 AND status = 'completed'
""")

tf = TurboFrame(spark_df)

result = (
    tf.groupby(["region", "category"])
      .agg({"amount": "sum", "quantity": "mean"})
      .sort("amount_sum", ascending=False)
)

# Back to Spark for writing
spark_result = result.to_spark(spark)
spark_result.write.format("delta").mode("overwrite").saveAsTable("summary")

Example 3: CSV Analysis

tf = TurboFrame.read_csv("exports/sales_2024.csv", columns=["region", "amount", "product"])

top_regions = (
    tf.filter("amount > 500")
      .groupby("region")
      .agg({"amount": "sum", "product": "count"})
      .rename({"amount_sum": "total_sales", "product_count": "num_orders"})
      .sort("total_sales", ascending=False)
)

top_regions.show()
top_regions.to_csv("reports/top_regions.csv")

Example 4: Join + Aggregate

orders = TurboFrame.read_delta("/Tables/orders", columns=["order_id", "customer_id", "amount"])
customers = TurboFrame.read_delta("/Tables/customers", columns=["customer_id", "segment"])

result = (
    orders.join(customers, on="customer_id", how="left")
          .groupby("segment")
          .agg({"amount": "sum", "order_id": "count"})
          .sort("amount_sum", ascending=False)
)

result.show()

Example 5: Process Huge Table in Batches

# 100M row table — never loads more than 200K rows at a time
results = TurboFrame.read_delta_batched(
    "/Tables/all_transactions",
    batch_fn=lambda b: b.groupby("region").agg({"amount": "sum", "qty": "count"}),
    columns=["region", "amount", "qty"],
    filters=[("year", ">=", 2023)],
    batch_size=200_000
)

# Final merge of partial results
final = results.groupby("region").agg({"amount_sum": "sum", "qty_count": "sum"})
final.show()

---

Fabric Notebook Setup

Option 1: Install from PyPI (Recommended)

# Cell 1
%pip install turboframe[delta]

# Cell 2
from turboframe import TurboFrame
tf = TurboFrame.read_delta("/lakehouse/default/Tables/your_table")

Option 2: Fabric Environment (Persistent — no reinstall each session)

1. Workspace -> + New -> Environment
2. Public Libraries -> Add from PyPI -> turboframe[delta]
3. Click Publish
4. In Notebook -> Environment dropdown -> select your environment

Option 3: From .whl file

# Upload turboframe-2.1.0-py3-none-any.whl to lakehouse Files/
%pip install /lakehouse/default/Files/turboframe-2.1.0-py3-none-any.whl

---

Performance Tips

1. Always use columns= when reading — loading 3 columns instead of 50 is 10x faster
2. Use filters= in read_delta() — skips entire Parquet row groups at the storage level
3. Filter early — apply .filter() before .groupby() to reduce data volume
4. For huge tables — use read_delta_batched() or let Spark pre-filter first
5. Use row_limit= as a safety net against OOM
6. Worker count — auto-detects CPU cores. Override with max_workers=8 if needed

---

Version

Current: 2.1.0

License

MIT