data load tool (dlt) —— 开源 Python 库，自动化处理所有繁琐的数据加载任务

无论是在 Google Colab 笔记本、AWS Lambda 函数、Airflow DAG、本地电脑，
还是 AI 编码代理中——dlt 都可以轻松集成。

🚀 加入我们充满活力的开发者社区，共同构建未来！

安装

dlt 支持 Python 3.10 至 Python 3.14。请注意，部分可选扩展包尚未支持 Python 3.14，因此该版本的支持被视为实验性功能。

pip install dlt

根据需要添加数据源和目标数据库所需的扩展包，例如：

pip install "dlt[duckdb]"        # 本地 DuckDB 目标
pip install "dlt[bigquery]"      # 或 snowflake、postgres、redshift、databricks、athena 等
pip install "dlt[s3]"            # 或 gs、az（云文件系统）
pip install "dlt[sql_database]"  # 从任意 SQL 数据库读取
pip install "dlt[hub]"           # 数据质量、转换和 AI 功能（见下文）

喜欢用 uv？试试 uv add "dlt[duckdb]"。

快速开始

以声明方式描述一个 API 并将其加载到 DuckDB——dlt 会为你处理请求、分页、模式推断和类型转换：

import dlt
from dlt.sources.rest_api import rest_api_source

# 1. 以声明方式描述 API
source = rest_api_source({
    "client": {"base_url": "https://pokeapi.co/api/v2/"},
    "resources": [
        {"name": "pokemon", "endpoint": {"path": "pokemon", "params": {"limit": 1000}}},
    ],
})

# 2. 将管道指向任意目标
pipeline = dlt.pipeline(
    pipeline_name="pokemon",
    destination="duckdb",
    dataset_name="pokemon_data",
)

# 3. 提取、规范化并加载
print(pipeline.run(source))

# 4. ...然后直接读回为 DataFrame
print(pipeline.dataset().pokemon.df())

...或者加载任何 Python 可迭代对象——一个 资源 就是一个生成器，dlt 会推断模式、为列设置类型并写入表：

import dlt

@dlt.resource(table_name="players", primary_key="id", write_disposition="merge")
def players():
    yield {"id": 1, "name": "Magnus", "rating": 2839}
    yield {"id": 2, "name": "Pragg", "rating": 2758}

dlt.pipeline(destination="duckdb", dataset_name="chess").run(players())

查看 Colab 中的超简单演示，或更高级的 Hugging Face + Marimo 笔记本演示。

为什么选择 dlt

dlt 能将杂乱、通常非结构化的数据源加载为结构良好、类型明确的数据集。它是一个库，而非平台——你可以通过 pip install 将其安装到现有代码中，并保留现有的工作流程和工具。没有黑盒：干净的 Pythonic 接口、可读的文件格式、可检查的模式、无隐藏副作用。

dlt 及其文档为 LLM 和编码代理而生。将下面的类型化、声明式原语与 dlthub.com/context 和 LLM 原生工作流 结合，即可从提示词直达可运行的管道——覆盖 5000+ 数据源——通常一次即可完成。

从任意数据源提取

REST APIs —— 以声明方式描述端点；在源头进行过滤、映射和扁平化记录（文档）：

from dlt.sources.rest_api import rest_api_source

source = rest_api_source({
    "client": {
        "base_url": "https://api.example.com/v1",
        "paginator": {"type": "cursor", "cursor_path": "next_cursor"},
    },
    "resources": [
        {
            "name": "guests",
            "endpoint": {"path": "events/guests"},
            "processing_steps": [
                {"filter": lambda r: r["approval_status"] == "approved"},
                {"map": lambda r: {**r, "email": r["email"].lower()}},
            ],
        },
    ],
})

SQL 数据库 —— 直接从数据库反射表和类型（文档）：

from dlt.sources.sql_database import sql_database

source = sql_database("mysql+pymysql://user:pass@host/db")

任意存储桶中的文件 —— 列出文件，然后从本地磁盘、S3、GCS 或 Azure 解析 CSV / JSONL / Parquet（文档）：

from dlt.sources.filesystem import filesystem, read_csv_duckdb

source = (
    filesystem(bucket_url="s3://my-bucket/data", file_glob="*.csv")
    | read_csv_duckdb()
).with_name("events")

DataFrames 和 Arrow —— pandas、Polars 和 Arrow 表可直接加载；Arrow 支持的表可零拷贝移动：

import dlt
import pandas as pd

df = pd.DataFrame({"event": ["dlt summit", "DuckCon"], "signups": [1240, 860]})
dlt.pipeline(destination="duckdb", dataset_name="events").run(df, table_name="events")

查看更多 数据源。

加载到 20+ 个目标——只需切换一个字符串

同一个资源可在任何地方运行。修改 destination 字符串，dlt 会自动处理凭据、目标方言的 DDL、暂存和模式漂移：

pipeline = dlt.pipeline(
    pipeline_name="luma",
    destination="duckdb",       # → snowflake, bigquery, postgres, redshift, databricks,
    dataset_name="luma_data",   #   athena, clickhouse, motherduck, filesystem (S3/GCS/Azure),
)                               #   iceberg, delta, ... 以及自定义逆向 ETL 目标
pipeline.run(source)

dlt 处理你不想操心的部分：

• 凭据 → secrets.toml / 环境变量，自动注入
• DDL → 使用目标方言的 CREATE TABLE
• 类型映射 → 将源类型转换为目标类型
• 暂存 → 为需要暂存的数仓使用 S3 / GCS
• 模式漂移 → 动态执行 ALTER TABLE

浏览所有 支持的目标，或构建 自定义目标。

用装饰器声明意图

装饰器让你声明想要什么——增量加载、合并策略、模式契约、列提示——而不是手动实现。每个选项都可以在运行时覆盖（文档）：

import dlt

@dlt.resource(
    primary_key="id",
    write_disposition="merge",                       # 按主键执行 upsert
    columns={"email": {"x-annotation-pii": True}},   # 设置列类型和注解
    schema_contract={"columns": "freeze"},           # 拒绝意外列
)
def events(
    updated_at=dlt.sources.incremental("updated_at"),  # 仅加载新增/变更行
):
    yield from fetch_events(since=updated_at.last_value)


@dlt.source
def luma(api_key: str = dlt.secrets.value):
    return events(), guests()   # 将多个资源分组到共享配置/认证下

模式契约 在入口强制数据的形状，提供三种模式——evolve（接受并适配模式）、freeze（拒绝该记录）和 discard（丢弃问题行/列）——分别应用于 tables、columns 和 data_type。你还可开箱即用地获得 模式推断、嵌套数据规范化、增量加载 和 密钥与配置注入。

读回数据：Dataset API

管道是持久化的。使用 dlt.attach 按名称重新连接，并以适合你工具的形状读取任意表（文档）：

import dlt

pipeline = dlt.attach(pipeline_name="luma", destination="duckdb", dataset_name="luma_data")

dataset = pipeline.dataset()
dataset.tables               # ['events', 'guests', ...]

guests = dataset.guests      # 惰性 dlt.Relation
guests.df()                  # pandas DataFrame
guests.arrow()               # pyarrow.Table（零拷贝）
guests.to_ibis()             # ibis 表达式——惰性、可组合

用 Ibis 转换——Python 输入，SQL 输出

将任意已加载的表提升为 Ibis 表达式，在 Python 中组合 group-by、join 和窗口函数，然后让 dlt 将其编译为目标方言的 SQL。只有在你请求结果时才实际执行：

import ibis

guests = pipeline.dataset().guests.to_ibis()

guests_by_event = (
    guests
    .group_by("event_id")
    .aggregate(n_guests=ibis._.api_id.count())
)

guests_by_event.to_pyarrow()   # 编译为 SQL 并在目标上执行

dlt 还支持 Python 和 SQL 数据访问、转换、管道检查 和 在 Marimo 笔记本中可视化数据。

文档

有关详细用法和配置，请参阅 官方文档。

示例

你可以在 examples 文件夹中找到各种用例示例，或在文档页面的 代码示例部分 中查看。

作为依赖添加

dlt 遵循 MAJOR.MINOR.PATCH 模式的语义化版本控制。

• major 表示破坏性变更和移除已弃用的功能
• minor 新功能，有时包含自动迁移
• patch 错误修复

建议使用兼容版本说明符仅允许 patch 级别更新。例如 dlt~=1.23.0 仅允许 >=1.23.0 且<1.24.0 的版本。

另请参阅我们的 发布说明，了解版本间的重要变更。

参与贡献

dlt 项目正在快速发展，我们很高兴你加入我们的社区！以下是参与方式：

• 与社区联系：加入我们的 Slack，与其他 dlt 用户和贡献者交流
• 报告问题和建议功能：请使用 GitHub Issues 报告错误或建议新功能。创建新 issue 前，请搜索可能存在的重复项，如有请添加评论
• 跟踪我们的工作进度和计划：查看我们的 公开 GitHub 项目
• 改进文档：帮助我们完善 dlt 文档

贡献代码

在提交 PR 前，请阅读 CONTRIBUTING。

• 📣 新增目标数据库不太可能被合并，因为维护成本较高（但欢迎改进 SQLAlchemy 目标以支持更多方言）
• 重要变更需要测试和文档，且在许多情况下编写测试会比编写代码更费时
• 欢迎提交错误修复和改进！你将获得测试和文档编写的帮助，以及专业的代码审查

赞助商

Blacksmith 是 GitHub 托管运行器的即插即用替代品，可将我们的 CI/CD 管道速度提升 2 倍，成本降低高达 75%。我们感谢 Blacksmith 提供的免费 CI/CD 分钟数赞助，这帮助我们保持构建速度和降低成本。

许可证

dlt 以 Apache 2.0 许可证 发布。