dlt-skill

dlt Pipeline Creator

使用以下决策树选择管道类型，然后按照核心工作流操作。

快速开始： 1）使用决策树。2）按照核心工作流操作。3）根据需要使用模式和参考资料。

管道类型决策树

当用户请求创建 dlt 管道时，确定要创建的类型：

START: User wants to create a dlt pipeline
│
├─→ Is there a dlt verified source available for this platform?
│   (Check: https://dlthub.com/docs/dlt-ecosystem/verified-sources)
│   │
│   YES → Use VERIFIED SOURCE approach
│   │     Examples: Salesforce, GitHub, Stripe, HubSpot, Slack
│   │     Action: Guide user through `dlt init <source> <destination>`
│   │
│   NO → Continue to next question
│
├─→ Is this a REST API with standard patterns?
│   (Standard auth, pagination, JSON responses)
│   │
│   YES → Use DECLARATIVE REST API approach
│   │     Examples: Pokemon API, simple REST APIs with clear endpoints
│   │     Action: Create config-based pipeline with rest_api_source
│   │
│   NO → Continue to next question
│
└─→ Does this require custom logic or Python packages?
    │
    YES → Use CUSTOM PYTHON approach
          Examples: Python packages (simple-salesforce), complex transformations,
                   non-standard APIs, custom data sources
          Action: Create custom source with @dlt.source and @dlt.resource decorators

核心工作流

1. 了解需求

提出澄清问题：

• 源：数据源是什么？（API URL、平台名称、数据库等）
• 源类型：是否匹配已验证的数据源、REST API，还是需要自定义代码？
• 目标位置：应将数据加载到哪里？（DuckDB、BigQuery、Snowflake 等）
• 资源：需要哪些具体数据/端点？
• 增量加载：管道应增量加载数据，还是执行全量刷新？
• 身份验证：需要哪些凭据？

2. 选择管道方案

根据上述决策树，选择：

• 已验证的数据源 - 预构建且经过测试的连接器
• 声明式 REST API - 基于配置的 REST API 管道
• 自定义 Python - 使用 Python 代码进行完全控制

3. 初始化或创建管道

已验证的数据源

dlt init <source_name> <destination_name>

示例：

• dlt init salesforce bigquery
• dlt init github duckdb
• dlt init stripe snowflake

声明式 REST API 或自定义 Python

使用此技能中的模板 assets/templates/（如有需要，将其复制到项目中）：

• declarative_rest_pipeline.py - 用于 REST API
• custom_python_pipeline.py - 用于自定义数据源

4. 安装所需软件包

**推荐：**使用辅助脚本（可检测 pip/uv/poetry）：

python scripts/install_packages.py --destination <destination_name>

手动安装：pip install "dlt[<destination>,workspace]"（例如 bigquery、snowflake）。对于 DuckDB，只需使用 dlt[workspace]。workspace 附加项是运行 dlt pipeline <name> show 和仪表板所必需的。

5. 配置凭据

创建或更新 .dlt/secrets.toml：

结构：

[sources.<source_name>]
# Source credentials here

[destination.<destination_name>]
# Destination credentials here

使用模板：assets/templates/.dlt/secrets.toml

重要：提醒用户将 .dlt/secrets.toml 添加到 .gitignore！

DuckDB 说明：DuckDB 不需要在 secrets.toml 中配置凭据。只需在管道或 config.toml 中指定数据库文件路径。

6. 配置管道设置

创建或更新 .dlt/config.toml，用于存放非敏感设置：

[sources.<source_name>]
base_url = "https://api.example.com"
timeout = 30

[destination.<destination_name>]
location = "US"

使用模板：assets/templates/.dlt/config.toml

7. 实现管道逻辑

根据需求完善管道代码：

对于已验证的数据源：

• 使用 .with_resources() 自定义资源选择
• 使用 .apply_hints() 配置增量加载
• 参见：references/verified-sources.md

对于声明式 REST API：

• 定义客户端配置（base_url、auth）
• 配置资源和端点
• 设置分页和增量加载
• 资源级选项（例如 max_table_nesting、table_name）在配置中的资源字典内设置；参见 references/rest-api-source.md 中的资源配置。
• 参见：references/rest-api-source.md

对于自定义 Python：

• 实现 @dlt.source 和 @dlt.resource 函数
• 使用生成器和 yield 模式
• 配置写入策略和主键
• 参见：references/custom-sources.md

8. 配置增量加载（如有需要）

对于只应加载新增/已更改数据的管道：

• 确定游标字段（时间戳、ID）
• 将写入策略设为 merge
• 定义用于去重的主键
• 配置增量参数

参见：references/incremental-loading.md

9. 测试并运行管道

python <pipeline_file>.py

检查是否有错误，并确认数据已正确加载。

10. 检查结果

前置条件：确保已安装 dlt[workspace]（使用 install_packages.py 时默认包含）。

打开 dlt 仪表板，检查已加载的数据：

dlt pipeline <pipeline_name> show

或者使用辅助脚本：

python scripts/open_dashboard.py <pipeline_name>

管道模式

模式 1：已验证的数据源 — 选择特定资源

from salesforce import salesforce_source

source = salesforce_source()
pipeline = dlt.pipeline(
    pipeline_name='salesforce_pipeline',
    destination='bigquery',
    dataset_name='salesforce_data'
)

# Load only specific Salesforce objects
pipeline.run(source.with_resources("Account", "Opportunity", "Contact"))

模式 2：声明式 REST API — 简单端点

from dlt.sources.rest_api import rest_api_source

config = {
    "client": {
        "base_url": "https://pokeapi.co/api/v2/",
    },
    "resources": [
        "pokemon",
        {
            "name": "pokemon_details",
            "endpoint": "pokemon/{name}",
            "write_disposition": "merge",
            "primary_key": "id"
        }
    ]
}

pipeline = dlt.pipeline(
    pipeline_name="pokemon",
    destination="duckdb",
    dataset_name="pokemon_data"
)
pipeline.run(rest_api_source(config))

模式 3：自定义 Python — 使用 Python 包

import dlt
from simple_salesforce import Salesforce

@dlt.source
def salesforce_custom(username=dlt.secrets.value, password=dlt.secrets.value):
    sf = Salesforce(username=username, password=password)

    @dlt.resource(write_disposition='merge', primary_key='Id')
    def accounts():
        records = sf.query_all("SELECT Id, Name FROM Account")
        yield records['records']

    return accounts

pipeline = dlt.pipeline(
    pipeline_name='salesforce_custom',
    destination='duckdb',
    dataset_name='salesforce'
)
pipeline.run(salesforce_custom())

模式 4：使用 REST API 增量加载

config = {
    "client": {
        "base_url": "https://api.github.com/repos/dlt-hub/dlt/",
        "auth": {"token": dlt.secrets["github_token"]}
    },
    "resources": [
        {
            "name": "issues",
            "endpoint": {
                "path": "issues",
                "params": {
                    "state": "all",
                    "since": "{incremental.start_value}"
                }
            },
            "incremental": {
                "cursor_path": "updated_at",
                "initial_value": "2024-01-01T00:00:00Z"
            },
            "write_disposition": "merge",
            "primary_key": "id"
        }
    ]
}

模式 5：REST API 数据源的非端点资源（例如，由数据库或文件提供的参数）

使用非端点资源（例如，由数据库或文件提供的参数），从数据库、文件或其他非 API 数据源驱动 REST API 调用。在 dlt 管道上下文之外预先获取数据，以避免 dlt.attach() / 上下文冲突。种子资源必须生成一个列表，其中包含多个字典，以便每一行都驱动一次 API 请求。

import duckdb
import dlt
from dlt.sources.rest_api import rest_api_source

# 1. Pre-fetch data from database (outside dlt context)
def get_locations():
    conn = duckdb.connect("locations.duckdb", read_only=True)
    result = conn.execute("SELECT id, lat, lng FROM locations").fetchall()
    conn.close()
    return [{"id": r[0], "lat": r[1], "lng": r[2]} for r in result]

# 2. Create seed resource
@dlt.resource(selected=False)
def locations():
    yield get_locations()  # Yield as LIST

# 3. Configure REST API with resolve
config = {
    "client": {"base_url": "https://api.weather.com/"},
    "resources": [
        locations(),
        {
            "name": "weather",
            "endpoint": {
                "path": "forecast",
                "params": {
                    "lat": "{resources.locations.lat}",
                    "lng": "{resources.locations.lng}"
                },
                "data_selector": "$",
                "paginator": "single_page"
            },
            "include_from_parent": ["id"],
            "primary_key": "_locations_id"
        }
    ]
}

source = rest_api_source(config)
pipeline = dlt.pipeline(
    pipeline_name="weather",
    destination="duckdb",
    dataset_name="weather_data"
)
pipeline.run(source)

参见：references/rest-api-source.md（非 REST 端点资源、查询参数/路径参数、单对象响应、include_from_parent）。

最佳实践（数据工程）

• 密钥：使用 .dlt/secrets.toml；切勿硬编码；将其添加到 .gitignore
• 主键：为合并操作和去重设置主键
• 写入策略：append（事件）、merge（有状态数据）、replace（快照）
• 性能：生成页面而不是行；尽可能使用增量加载

更多信息请参见 references/performance-tuning.md、references/incremental-loading.md 和 references/troubleshooting.md。

常见问题与解决方案

身份验证（OAuth2）： 在 REST 配置中使用 "auth": {"type": "oauth2_client_credentials", ...}。对于自定义 Python，使用 dlt.sources.helpers.rest_client.auth.OAuth2ClientCredentials 和 paginate()。参见 references/rest-api-source.md。

自定义分页 / 嵌套数据 / 性能： 参见 references/rest-api-source.md、references/custom-sources.md、references/performance-tuning.md。

参考文档 — 何时阅读哪些内容

• 完整工作流 / 分步示例 → references/examples.md
• 已验证的数据源 → references/verified-sources.md
• 声明式 REST API → references/rest-api-source.md
• 自定义 Python 数据源 → references/custom-sources.md
• 增量加载 → references/incremental-loading.md
• 性能 → references/performance-tuning.md
• 错误 / 调试 → references/troubleshooting.md
• dlt 基础知识 → references/core-concepts.md

模板和脚本

模板（assets/templates/）

• custom_python_pipeline.py - 自定义 Python 管道骨架
• verified_source_pipeline.py - 已验证的数据源管道骨架
• declarative_rest_pipeline.py - 声明式 REST API 管道骨架
• .dlt/config.toml - 配置文件模板
• .dlt/secrets.toml - 密钥文件模板
• .gitignore - dlt 项目的 Git 忽略模板

脚本（scripts/）

• install_packages.py - 安装 dlt + 目标端附加组件（包括 workspace）。在设置新项目或添加目标端时运行。
• open_dashboard.py - 打开管道仪表板（dlt pipeline <name> show）。运行管道后运行此脚本，以检查已加载的数据。

关键提醒

• 始终询问目标端——不要擅自假设
• 安全第一——切勿提交机密；使用 .dlt/secrets.toml 并提供 .gitignore
• 从简单开始——尽可能使用经过验证的数据源；逐步测试
• 阅读参考资料——仅在需要时加载详细文档