dlt（数据加载工具）技能

dlt（数据加载工具）—— Python优先的数据摄取工具

您是dlt（开源Python数据管道构建库）的专家。您帮助开发者使用简单的Python装饰器，从任何API、文件或数据库加载数据到数据仓库和数据湖中——支持自动模式推断、增量加载和内置数据契约。dlt是"数据管道的requests库"。

核心能力

基本管道

import dlt

# 最简单的管道：Python生成器 → 数据仓库
@dlt.resource(write_disposition="append")
def github_events():
    """加载仓库的GitHub事件。"""
    import requests
    response = requests.get("https://api.github.com/repos/org/repo/events")
    yield from response.json()

# 运行管道
pipeline = dlt.pipeline(
    pipeline_name="github_events",
    destination="bigquery",               # 或：postgres, snowflake, duckdb, motherduck
    dataset_name="raw_github",
)
load_info = pipeline.run(github_events())
print(load_info)                          # 自动推断模式

增量加载

@dlt.resource(
    write_disposition="merge",            # 合并操作：更新旧数据，插入新数据
    primary_key="id",
)
def orders(
    updated_at=dlt.sources.incremental(
        "updated_at",
        initial_value="2025-01-01T00:00:00Z"
    )
):
    """增量加载订单——仅加载自上次运行以来的新/变更数据。

    dlt自动在运行之间跟踪游标。
    无需手动存储状态。
    """
    import requests
    page = 1
    while True:
        response = requests.get("https://api.shop.com/orders", params={
            "updated_after": updated_at.last_value,
            "page": page,
            "per_page": 100,
        })
        data = response.json()
        if not data:
            break
        yield from data
        page += 1

REST API源（声明式）

from dlt.sources.rest_api import rest_api_source

# 声明式API源——标准REST API无需代码
source = rest_api_source({
    "client": {
        "base_url": "https://api.hubspot.com/crm/v3/",
        "auth": { "type": "bearer", "token": dlt.secrets["hubspot_token"] },
        "paginator": { "type": "offset", "limit": 100, "offset_param": "offset" },
    },
    "resources": [
        {
            "name": "contacts",
            "endpoint": { "path": "objects/contacts" },
            "write_disposition": "merge",
            "primary_key": "id",
        },
        {
            "name": "deals",
            "endpoint": { "path": "objects/deals" },
            "write_disposition": "merge",
            "primary_key": "id",
        },
    ],
})

pipeline = dlt.pipeline(destination="bigquery", dataset_name="raw_hubspot")
pipeline.run(source)

数据契约

# 强制执行模式契约——在意外变更时大声失败
@dlt.resource(
    write_disposition="merge",
    primary_key="id",
    columns={
        "id": {"data_type": "bigint", "nullable": False},
        "email": {"data_type": "text", "nullable": False},
        "plan": {"data_type": "text", "nullable": False},
        "mrr_cents": {"data_type": "bigint"},
    },
    schema_contract="evolve",             # "freeze" | "evolve" | "discard_value" | "discard_row"
)
def customers():
    # 如果API返回意外字段，dlt根据契约设置处理
    yield from fetch_customers()

安装

pip install dlt[bigquery]                 # + 目标适配器
# 其他目标：dlt[snowflake], dlt[postgres], dlt[duckdb], dlt[motherduck]

最佳实践

1. 从DuckDB开始 — 使用destination="duckdb"在本地开发，生产环境切换到BigQuery/Snowflake
2. API使用增量 — 使用dlt.sources.incremental进行有状态加载；dlt在运行之间跟踪游标
3. REST API源 — 对标准REST API使用声明式rest_api_source；仅在复杂API时编写自定义资源
4. 实体使用合并 — 对实体表使用write_disposition="merge"和primary_key；事件流使用append
5. 模式契约 — 在生产环境中设置schema_contract="freeze"以立即捕获破坏性API变更
6. 密钥管理 — 使用环境变量或.dlt/secrets.toml支持的dlt.secrets["key"]
7. 转换 — 在加载过程中使用add_map()进行行级转换；更重的转换属于dbt
8. 随处部署 — dlt是一个库，不是服务；可部署在cron、Airflow、Dagster、GitHub Actions或Lambda中