polars-etl-kit

High-performance ETL toolkit for Python — Polars + DuckDB powered.

Overview

polars-etl-kit is a collection of battle-tested ETL building blocks extracted from real-world financial data pipelines processing 1500+ Excel files per run. Each module is framework-agnostic and composable — use one or use them all.

Modules

Install

# Core (polars + duckdb)
pip install polars-etl-kit

# With Excel support
pip install polars-etl-kit[excel]

# With Kedro integration
pip install polars-etl-kit[kedro]

# With geocoding
pip install polars-etl-kit[geo]

# Everything
pip install polars-etl-kit[all]

Quick start

Parallel Excel processing

from polars_etl_kit.excel import ExcelProcessor, FileCache, scan_files

# Scan for files
files = scan_files("data/01_raw", patterns=[r"(?P<code>\w+)_(?P<month>\d{6})\.xlsx"])

# Set up cache
cache = FileCache("cache/excel")

# Define your sheet handler
def handle_sheet(sheet_name, df, file_meta):
    # Your parsing logic here
    return df.with_columns(pl.lit(file_meta["code"]).alias("source_code"))

# Process in parallel
processor = ExcelProcessor(max_workers=8, file_cache=cache)
results = processor.process_files(
    files,
    process_sheet=handle_sheet,
    cache_suffixes=["parsed"],
)

Schema registry + DuckDB warehouse

from polars_etl_kit import SchemaRegistry, WarehouseBuilder

# Load schema from YAML
registry = SchemaRegistry("schemas/dimensions.yaml", "schemas/facts.yaml")

# Generate DDL
print(registry.generate_all_ddl())

# Build warehouse
wh = WarehouseBuilder("analytics.duckdb", schema="analytics")
wh.create_schema()
wh.load_dimension("dim_customer", "data/dim_customer.parquet", pk="customer_id")
wh.load_fact("fact_sales", "data/fact_sales.parquet", indexes=["customer_id", "date"])
df = wh.query("SELECT * FROM analytics.v_summary")

Name matching (with fuzzy fallback)

from polars_etl_kit.matching import NameMatcher

# enable_fuzzy=True: Levenshtein 4th-tier fallback for typos
matcher = NameMatcher("mappings/standard_names.yaml", enable_fuzzy=True)
result = matcher.match({
    "raw_text": "营业总收入",
    "clean_name": "营业总收人",       # 错字: "入"→"人"
    "category": "利润表",
})
# → {"code": "B001", "name": "营业收入", "fuzzy_score": 0.89, ...}

Schema inference

from polars_etl_kit.schema import infer_schema_from_polars
import polars as pl

df = pl.DataFrame({"id": [1, 2], "name": ["Alice", "Bob"], "score": [95.0, 82.5]})
schema = infer_schema_from_polars(df, "dim_students", "学生维度")
print(schema.generate_duckdb_ddl())
# → CREATE TABLE IF NOT EXISTS dim_students (
#     id BIGINT NOT NULL, name VARCHAR(255), score DOUBLE);

Tree building

from polars_etl_kit.tree import build_tree, find_parent_by_prefix
import polars as pl

nodes = pl.DataFrame([
    {"node_id": "A_9", "parent_id": "#", "node_name": "Group"},
    {"node_id": "A_1", "parent_id": "A_9", "node_name": "Sub A"},
    {"node_id": "B_9", "parent_id": "A_9", "node_name": "Sub B"},
])

tree = build_tree(nodes, node_id_col="node_id", parent_id_col="parent_id",
                  node_name_col="node_name", find_parent=find_parent_by_prefix)

Multi-provider geocoding

from polars_etl_kit.geo import BatchGeocoder, get_geocoder
import polars as pl

entities = pl.DataFrame({
    "code": ["E001"], "suffix": ["9"], "address": ["北京市朝阳区"],
})

# Switch providers: "amap" / "baidu" / "tencent"
geocoder = BatchGeocoder(
    api_key="your_key",
    cache_path="geo_cache.parquet",
    provider="baidu",     # 百度地图
    key_cols=["code", "suffix"],
)
results = geocoder.run(entities, address_col="address")

# Or use factory for single queries
client = get_geocoder("tencent", api_key="your_key")  # 腾讯地图
# async with client as c:
#     result = await c.geocode("北京市朝阳区")

Philosophy

• Polars-first: all DataFrame operations use Polars for speed
• Framework-agnostic: modules work standalone; Kedro is optional
• Composable: each module has minimal internal dependencies
• Battle-tested: extracted from production pipelines processing 1500+ files
• Zero hardcoded paths: everything is parameterized

License

MIT — see LICENSE.