data-engineering-core（数据工程核心）

Core Data Engineering

Use this skill for day-to-day Python data engineering decisions: dataframe processing, embedded OLAP SQL, Arrow interchange, ETL structure, and resilience patterns.

When to use this skill

Use this skill when the task involves one or more of:

• Polars transformations and lazy query optimization
• DuckDB SQL analytics, MERGE/upsert, or Parquet querying
• PyArrow table/dataset interchange and Parquet scans
• Python ETL pipeline structure (extract/transform/load, logging, retries, idempotency)
• PostgreSQL-to-lake/warehouse ingestion patterns

If the task is primarily about cloud storage auth/connectors, use:

• @data-engineering-storage-remote-access
• @data-engineering-storage-authentication

If the task is primarily about ACID lakehouse table behavior, use:

• @data-engineering-storage-lakehouse

Quick tool selection

Rule of thumb:

1. Start transformations in Polars lazy.
2. Use DuckDB for heavy SQL/windowing/joins over files.
3. Keep boundaries in Arrow/Parquet.

Core implementation rules

1) Prefer lazy execution

• Use pl.scan_* over pl.read_* for large inputs.
• Chain filters/projections before collect().
• Avoid row-wise loops.

2) Push work down

• Push filtering into file scans (predicate pushdown).
• Select only needed columns (column pruning).
• Partition data by query dimensions (typically date/tenant/region).

3) Keep writes idempotent

• Prefer MERGE/upsert semantics where possible.
• If append-only, track watermark/checkpoints.
• Ensure retries do not duplicate side effects.

4) Protect boundaries

• Use parameterized SQL for values.
• Treat dynamic identifiers (table/column names) separately and validate.
• Never hardcode secrets.

5) Instrument and validate

• Log row counts and stage durations.
• Validate schema/required columns at stage boundaries.
• Record checkpoints/watermarks for incremental flows.

Minimal safe ETL shape

import polars as pl
import duckdb

def run_etl(source_path: str, target_table: str) -> None:
    lazy = (
        pl.scan_parquet(source_path)
        .filter(pl.col("value").is_not_null())
        .select(["id", "event_ts", "value", "category"])
    )

    df = lazy.collect()

    with duckdb.connect("analytics.db") as con:
        con.sql("CREATE OR REPLACE TABLE staging AS SELECT * FROM df")
        con.sql(f"""
            CREATE TABLE IF NOT EXISTS {target_table} AS
            SELECT * FROM staging WHERE 1=0
        """)
        con.sql(f"INSERT INTO {target_table} SELECT * FROM staging")

For production-grade structure, use:

• templates/complete_etl_pipeline.py

Progressive disclosure (read next as needed)

• patterns/etl.md — canonical ETL pipeline structure
• patterns/incremental.md — watermark/CDC/incremental loading patterns
• templates/complete_etl_pipeline.py — full template with logging and checkpoints
• core-detailed.md — comprehensive reference (extended examples)

Related skills

• @data-engineering-storage-lakehouse — Delta/Iceberg/Hudi behavior
• @data-engineering-storage-remote-access — fsspec/pyarrow.fs/obstore cloud access
• @data-engineering-orchestration — Prefect/Dagster/dbt orchestration
• @data-engineering-quality — Pandera / Great Expectations validation
• @data-engineering-observability — OTel + Prometheus monitoring
• @data-engineering-ai-ml — embedding/vector/RAG pipelines

References

• Polars Documentation
• DuckDB Documentation
• PyArrow Documentation
• psycopg Documentation