polars-etl-kit 0.1.0

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

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

Module	What it does	Core deps
excel	Parallel Excel parser with MD5 caching	polars, fastexcel
schema	YAML-driven schema registry → DDL generation	pyyaml
warehouse	DuckDB star-schema builder (dim/fact)	duckdb, polars
matching	Multi-tier name matching engine	pyyaml
tree	Hierarchical tree from flat data	polars
geo	AMap (高德) batch geocoder with cache	aiohttp, polars
db	DuckDB connection with retry	duckdb
kedro	Custom Kedro datasets & dynamic hooks	kedro

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

from polars_etl_kit.matching import NameMatcher

matcher = NameMatcher("mappings/standard_names.yaml")
result = matcher.match({
    "raw_text": "营业总收入",
    "clean_name": "营业总收入",
    "category": "利润表",
})
# → {"code": "B001", "name": "营业收入", "path": "利润表 > 营业收入", ...}

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)

AMap geocoding

from polars_etl_kit.geo import BatchGeocoder
import polars as pl

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

geocoder = BatchGeocoder(
    api_key="your_amap_key",
    cache_path="geo_cache.parquet",
    key_cols=["code", "suffix"],
)
results = geocoder.run(entities, address_col="address",
                       fallback_cols=["province", "city", "area"])

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.