data-pipelines

你正在构建数据管道。通用模式是 摄取（获取数据）→ 转换（清理、建模、连接）→ 查询（分析）→ 探索（笔记本、应用、可视化）。

每个步骤使用的具体工具取决于项目。推荐的默认选择：

步骤	首选工具	替代方案
摄取	dlt	纯 Python 脚本、shell + curl
转换	sqlmesh	纯 SQL 脚本、dbt、Python 脚本
查询引擎	DuckDB / MotherDuck	—
DataFrame	polars	—
笔记本	marimo	—
项目管理	uv	—
可视化	altair、seaborn	—

语言偏好

优先使用 SQL（DuckDB 方言），其次是 Python，再其次是 bash。使用能完成任务的最简单语言。

与其他技能的交叉引用

本技能介绍摄取、转换和笔记本工作的主观推荐技术栈。底层工具的使用请参考：

• DuckDB 语法（Friendly SQL、文件读取、嵌套类型）→ motherduck-duckdb-sql 或 duckdb-docs
• MotherDuck 连接、探索、共享、Dives → motherduck-* 技能
• marimo 笔记本（单元格、响应式、SQL 单元格）→ marimo-notebook
• 本地 DuckDB 会话管理 → attach-db、query、read-file

项目布局

数据项目应采用以下结构：

project/
├── ingest/              # Extraction and loading scripts
│   ├── <source>.py      # One file per data source
│   └── .dlt/            # dlt config (if using dlt)
├── transform/           # Transformation logic
│   ├── models/          # SQL models (sqlmesh or plain SQL)
│   └── config.yaml      # sqlmesh config (if using sqlmesh)
├── notebooks/           # Exploration and analysis
│   └── *.py             # marimo notebooks (plain .py files)
├── data/                # Local data files (gitignored)
├── pyproject.toml       # Dependencies
├── uv.lock              # Locked dependencies (committed)
└── *.duckdb             # Local database (gitignored)

并非每个项目都需要所有目录——简单的分析可能只需要 notebooks/ 和一个 DuckDB 文件。按需扩展。

uv — 项目管理

切勿直接使用 pip。所有 Python 工作都通过 uv 进行。

uv init my-project                    # New project
uv add polars duckdb                  # Add dependencies
uv sync                               # Install into .venv
uv run python script.py               # Run in project venv
uv run --with requests script.py      # Ad-hoc dependency

内联脚本依赖项（PEP 723），适用于独立脚本：

# /// script
# dependencies = ["dlt[duckdb]", "polars"]
# requires-python = ">=3.12"
# ///

使用 uv run script.py 运行——依赖项会自动解析。

始终提交 uv.lock。使用 pyproject.toml 声明依赖项，切勿使用 requirements.txt。

polars — DataFrame

需要使用 Python 逻辑时使用 polars，例如复杂的字符串转换、机器学习特征和行级条件判断。对于连接、聚合和窗口函数，优先使用 SQL。

关键模式

import polars as pl

# Lazy evaluation (always prefer for production)
lf = pl.scan_parquet("events/*.parquet")
result = (
    lf.filter(pl.col("event_date") >= "2024-01-01")
    .group_by("user_id")
    .agg(pl.col("amount").sum().alias("total_spend"))
    .sort("total_spend", descending=True)
    .collect()
)

# Three contexts
df.select(...)         # Pick/transform columns (output has ONLY these)
df.with_columns(...)   # Add/overwrite columns (keeps all originals)
df.filter(...)         # Keep rows matching condition

DuckDB 互操作（通过 Arrow 实现零拷贝）：

import duckdb
result = duckdb.sql("SELECT * FROM df WHERE amount > 100").pl()

dlt — 数据摄取

当项目使用 dlt 进行数据摄取时，可用它处理 API 调用、分页、模式推断、增量加载和状态管理。

搭建并运行

dlt init rest_api duckdb             # Scaffold pipeline
uv run python pipeline.py            # Run extraction
dlt pipeline <name> info             # Inspect state
dlt pipeline <name> schema           # View inferred schema

管道模式

最小管道：

import dlt

pipeline = dlt.pipeline(
    pipeline_name="my_pipeline",
    destination="duckdb",
    dataset_name="raw",
)
info = pipeline.run(data, table_name="events")

增量加载：

@dlt.resource(write_disposition="merge", primary_key="id")
def users(updated_at=dlt.sources.incremental("updated_at")):
    yield from fetch_users(since=updated_at.last_value)

REST API 源（声明式）：

from dlt.sources.rest_api import rest_api_source

source = rest_api_source({
    "client": {"base_url": "https://api.example.com/v1"},
    "resource_defaults": {"primary_key": "id", "write_disposition": "merge"},
    "resources": [
        "users",
        {
            "name": "events",
            "write_disposition": "append",
            "endpoint": {
                "path": "events",
                "incremental": {"cursor_path": "created_at", "initial_value": "2024-01-01"},
            },
        },
    ],
})

写入策略：

策略	行为	适用场景
append	插入行（默认）	不可变事件、日志
replace	删除并重建	小型查找表
merge	按 primary_key 执行 Upsert	可变记录

目标端：duckdb（本地文件）、motherduck（云端）。设置 motherduck_token 环境变量，或在 .dlt/secrets.toml 中配置。

sqlmesh — 转换

当项目使用 sqlmesh 进行转换时，它采用 SQL 优先的计划/应用工作流，不会意外更改生产环境。

搭建并运行

sqlmesh init duckdb                              # New project
sqlmesh init -t dlt --dlt-pipeline <name>        # From dlt schema
sqlmesh plan                                     # Preview + apply (dev)
sqlmesh plan prod                                # Promote to production
sqlmesh fetchdf "SELECT * FROM analytics.users"  # Ad-hoc query
sqlmesh test                                     # Run unit tests
sqlmesh ui                                       # Web interface

模型类型

类型	行为	适用场景
FULL	重写整个表	小型维度表
INCREMENTAL_BY_TIME_RANGE	处理新的时间区间	事实、事件、日志
INCREMENTAL_BY_UNIQUE_KEY	按键执行插入或更新	可变维度
SEED	静态 CSV 数据	参考/查找数据
VIEW	SQL 视图	简单透传
SCD_TYPE_2	缓慢变化维度	历史追踪

模型示例

MODEL (
    name analytics.stg_events,
    kind INCREMENTAL_BY_TIME_RANGE (time_column event_date),
    cron '@daily',
    grain (event_id),
    audits (NOT_NULL(columns=[event_id]))
);

SELECT
    event_id,
    user_id,
    event_type,
    event_date
FROM raw.events
WHERE event_date BETWEEN @start_date AND @end_date

配置（config.yaml）

gateways:
  local:
    connection:
      type: duckdb
      database: db.duckdb
default_gateway: local
model_defaults:
  dialect: duckdb

dlt 集成

sqlmesh init -t dlt 会根据 dlt 推断的架构自动生成外部模型和增量暂存模型。sqlmesh plan 会检测 dlt 中的架构变更。