/chdb-datastore

DataStore Examples

All examples are self-contained and runnable. Expected output is shown in comments.

Table of Contents

1. Pandas Replacement: One Import Change
2. Analyze Local Files
3. Cross-Source Join: MySQL + Parquet
4. Cross-Source Join: S3 + PostgreSQL
5. Three-Way Join: File + Database + Cloud
6. Data Lake Formats: Iceberg, Delta, Hudi
7. URI Shorthand Access
8. Cloud Storage Variants (S3/GCS/Azure/HDFS)
9. Cross-Source Write
10. Explore Remote Schema
11. Common Errors & Fixes

1. Pandas Replacement: One Import Change

The simplest way to use chdb — change one line, keep everything else:

# Before (standard pandas):
# import pandas as pd

# After (chdb-accelerated):
import chdb.datastore as pd

df = pd.DataStore({"name": ["Alice", "Bob", "Carol", "Dave"],
                   "dept": ["Eng", "Sales", "Eng", "Sales"],
                   "salary": [95000, 72000, 110000, 68000]})

# Same pandas API — everything works
result = (df[df["salary"] > 70000]
    .groupby("dept")
    .agg({"salary": ["mean", "count"]})
    .sort_values("mean", ascending=False))

print(result)
# Expected output:
#     dept  mean  count
# 0    Eng  102500    2
# 1  Sales   72000    1

Why it's faster: Operations compile to ClickHouse SQL and execute as a single optimized query, instead of step-by-step Python evaluation.

2. Analyze Local Files

from datastore import DataStore

# Parquet — pandas-style analysis
ds = DataStore.from_file("sales.parquet")
top_products = (ds[ds['revenue'] > 0]
    .groupby('product')
    .agg({'revenue': 'sum', 'quantity': 'sum'})
    .sort_values('revenue', ascending=False)
    .head(10))
print(top_products)

# CSV with filtering
ds = DataStore.from_file("employees.csv")
senior = ds[(ds['years'] > 5) & (ds['dept'] == 'Engineering')]
print(senior[['name', 'title', 'salary']].sort_values('salary', ascending=False))

# Glob pattern — query all matching files at once
ds = DataStore.from_file("logs/2024-*.csv")
errors = ds[ds['level'] == 'ERROR'].groupby('module')['message'].count()
print(errors.sort_values(ascending=False))

# See the SQL behind any query
print(top_products.to_sql())

3. Cross-Source Join: MySQL + Parquet

from datastore import DataStore

customers = DataStore.from_mysql(
    host="db:3306", database="crm", table="customers",
    user="reader", password="pass")

orders = DataStore.from_file("orders.parquet")

result = (customers
    .join(orders, left_on="id", right_on="customer_id", how="inner")
    .groupby("country")
    .agg({"amount": ["sum", "mean"], "order_id": "count"})
    .sort_values("sum", ascending=False))

print(result)
# Expected: country-level order summary with total, average, and count

print(result.to_sql())
# Shows the cross-source SQL generated by chdb

4. Cross-Source Join: S3 + PostgreSQL

from datastore import DataStore

events = DataStore.from_s3(
    "s3://analytics/events/2024-*.parquet",
    access_key_id="AKIA...", secret_access_key="secret...")

profiles = DataStore.from_postgresql(
    host="pg.example.com:5432", database="users",
    table="profiles", user="analyst", password="pass")

result = (events
    .join(profiles, left_on="user_id", right_on="id")
    .filter(events['event_type'] == 'purchase')
    .groupby(["country", "age_group"])
    .agg({"amount": "sum", "event_id": "count"})
    .sort_values("sum", ascending=False))

print(result)
# Expected: purchase events aggregated by country and age group

5. Three-Way Join: File + Database + Cloud

from datastore import DataStore

products = DataStore.from_file("products.csv")
orders = DataStore.from_mysql(
    host="db:3306", database="shop", table="orders",
    user="root", password="pass")
reviews = DataStore.from_s3("s3://feedback/reviews.parquet", nosign=True)

result = (orders
    .join(products, left_on="product_id", right_on="id")
    .join(reviews, left_on="product_id", right_on="product_id")
    .groupby("category")
    .agg({"amount": "sum", "rating": "mean", "review_id": "count"})
    .sort_values("sum", ascending=False))

print(result)
# Expected: category-level summary combining order amounts, review ratings, and counts

6. Data Lake Formats: Iceberg, Delta, Hudi

from datastore import DataStore

# Apache Iceberg on S3
ds = DataStore.from_iceberg(
    "s3://warehouse/iceberg/events",
    access_key_id="KEY", secret_access_key="SECRET")
print(ds.head(10))

# Delta Lake
ds = DataStore.from_delta(
    "s3://warehouse/delta/transactions",
    access_key_id="KEY", secret_access_key="SECRET")
summary = (ds.groupby("category")
    .agg({"amount": "sum"})
    .sort_values("sum", ascending=False))
print(summary)

# Hudi
ds = DataStore.from_hudi(
    "s3://warehouse/hudi/logs",
    access_key_id="KEY", secret_access_key="SECRET")
errors = ds[ds['level'] == 'ERROR']
print(errors.head(20))

7. URI Shorthand Access

from datastore import DataStore

# One-liner for any source
ds = DataStore.uri("sales.parquet")
ds = DataStore.uri("s3://public-data/dataset.parquet?nosign=true")
ds = DataStore.uri("mysql://root:pass@localhost:3306/shop/orders")
ds = DataStore.uri("postgresql://analyst:pass@pg:5432/analytics/events")
ds = DataStore.uri("clickhouse://ch:9440/analytics/hits?user=reader&password=pass")
ds = DataStore.uri("mongodb://user:pass@mongo:27017/logs.app_events")
ds = DataStore.uri("sqlite:///data/local.db?table=users")
ds = DataStore.uri("deltalake:///data/delta/events")

# After creating from any source, same pandas API
result = (ds[ds['value'] > 100]
    .groupby('category')
    .sum()
    .sort_values('value', ascending=False))
print(result)

8. Cloud Storage Variants

from datastore import DataStore

# AWS S3 (private)
ds = DataStore.from_s3("s3://my-bucket/data.parquet",
    access_key_id="AKIA...", secret_access_key="secret...")

# AWS S3 (public)
ds = DataStore.from_s3("s3://public-data/dataset.parquet", nosign=True)

# Google Cloud Storage
ds = DataStore.from_gcs("gs://my-bucket/data.parquet",
    hmac_key="KEY", hmac_secret="SECRET")
ds = DataStore.from_gcs("gs://public-bucket/data.parquet", nosign=True)

# Azure Blob Storage
ds = DataStore.from_azure(
    connection_string="DefaultEndpointsProtocol=https;AccountName=...;AccountKey=...",
    container="data", path="analytics/events.parquet")

# HDFS
ds = DataStore.from_hdfs("hdfs://namenode:9000/warehouse/events/*.parquet")

9. Cross-Source Write

from datastore import DataStore

# Read from MySQL, transform, write to Parquet
source = DataStore.from_mysql(
    host="db:3306", database="shop", table="orders",
    user="root", password="pass")
target = DataStore("file", path="output/orders_summary.parquet", format="Parquet")

target.insert_into("category", "total_revenue", "order_count").select_from(
    source
        .groupby("category")
        .select("category", "sum(amount) AS total_revenue", "count() AS order_count")
        .filter(source['amount'] > 0)
).execute()

# Read from S3, filter, write to local file
source = DataStore.from_s3("s3://logs/events.parquet", nosign=True)
target = DataStore("file", path="filtered_events.parquet", format="Parquet")

target.insert_into("user_id", "event_type", "ts").select_from(
    source.select("user_id", "event_type", "ts")
          .filter(source['event_type'] == 'error')
).execute()

10. Explore Remote Schema

from datastore import DataStore

# Connect to MySQL and browse the schema
mysql_ds = DataStore.from_mysql(
    host="db:3306", database="ecommerce",
    user="analyst", password="pass")

print(mysql_ds.databases())       # list all databases
print(mysql_ds.tables("ecommerce"))  # list tables in a database

# Quick preview of a specific table
orders = DataStore.from_mysql(
    host="db:3306", database="ecommerce",
    table="orders", user="analyst", password="pass")

print(orders.columns)      # → ['id', 'customer_id', 'amount', ...]
print(orders.dtypes)        # → {'id': 'UInt64', 'amount': 'Float64', ...}
print(orders.describe())    # → statistics for numeric columns
print(orders.head(5))       # → first 5 rows

11. Common Errors & Fixes

File not found

from datastore import DataStore

# Error: file not found
ds = DataStore.from_file("nonexistent.parquet")
# → Exception: FILE_NOT_FOUND

# Fix: check the path
import os
print(os.path.exists("nonexistent.parquet"))  # → False
ds = DataStore.from_file("data/sales.parquet")  # use correct path

Database host without port

# Error: connection timeout / refused
ds = DataStore.from_mysql(host="db", database="shop", table="orders",
    user="root", password="pass")
# → Connection refused

# Fix: include port in host string
ds = DataStore.from_mysql(host="db:3306", database="shop", table="orders",
    user="root", password="pass")

Join key type mismatch

# Error: join returns empty or wrong results
users = DataStore({"id": [1, 2, 3], "name": ["Alice", "Bob", "Carol"]})     # id is Int
orders = DataStore({"user_id": ["1", "2", "3"], "amount": [100, 200, 300]}) # user_id is String

result = users.join(orders, left_on="id", right_on="user_id")
print(result)  # → empty or incorrect

# Fix: ensure matching types — use .to_sql() to diagnose
print(result.to_sql())  # reveals the type mismatch in the JOIN condition
# Cast in source data, or use assign() to convert types before joining

Debugging with .to_sql()

from datastore import DataStore

ds = DataStore.from_file("sales.parquet")
result = (ds[ds['revenue'] > 1000]
    .groupby('product')
    .agg({'revenue': 'sum'})
    .sort_values('revenue', ascending=False)
    .head(10))

# See exactly what SQL will execute
print(result.to_sql())
# Output:
# SELECT "product", sum("revenue") AS "revenue"
# FROM file('sales.parquet', Parquet)
# WHERE "revenue" > 1000
# GROUP BY "product"
# ORDER BY "revenue" DESC
# LIMIT 10