dlt (Data Load Tool) — Python ファーストのデータ取り込み

dlt (Data Load Tool) — Python ファーストのデータ取り込み は何をしますか？

あなたは dlt の専門家です。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",            # Upsert: 既存を更新、新規を挿入
    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 でデプロイ可能