dlt-skill

dlt パイプラインクリエーター

以下の判断ツリーでパイプラインの種類を選び、続いてコアワークフローに従います。

クイックスタート： 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.attach() やコンテキストの競合を避けるため、dlt パイプラインのコンテキスト外でデータを事前に取得してください。シードリソースは、各行が 1 回の API リクエストを生成するように、辞書のリストをyieldする必要があります。

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（スナップショット）
• パフォーマンス：行ではなくページを yield してください。可能な場合は増分読み込みを使用してください。

詳しくは、references/performance-tuning.md、references/incremental-loading.md、references/troubleshooting.md を参照してください。

よくある課題と解決策

認証（OAuth2）： REST 設定では "auth": {"type": "oauth2_client_credentials", ...} を使用します。カスタム Python では、paginate() とともに dlt.sources.helpers.rest_client.auth.OAuth2ClientCredentials を使用します。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 ignore テンプレート

スクリプト（scripts/）

• install_packages.py - dlt と宛先の追加パッケージ（workspace を含む）をインストールします。新しいプロジェクトのセットアップ時や宛先の追加時に実行してください。
• open_dashboard.py - パイプラインのダッシュボード（dlt pipeline <name> show）を開きます。パイプラインの実行後に、読み込んだデータを確認するために実行してください。

重要な注意事項

• 宛先について必ず確認する - 決めつけないでください
• セキュリティを最優先する - シークレットをコミットしないでください。.dlt/secrets.toml を使用し、.gitignore を用意してください
• シンプルに始める - 利用可能な場合は検証済みのソースを使用し、段階的にテストしてください
• リファレンスを読む - 必要な場合にのみ詳細なドキュメントを読み込んでください