data-pipelines

データパイプラインを構築しています。一般的なパターンは、取り込み（データを取り込む）→ 変換（クレンジング、モデリング、結合）→ クエリ（分析）→ 探索（ノートブック、アプリ、可視化）です。

各ステップで使う具体的なツールは、プロジェクトによって異なります。推奨するデフォルトは次のとおりです。

ステップ	推奨ツール	代替手段
取り込み	dlt	通常の Python スクリプト、shell + curl
変換	sqlmesh	通常の SQL スクリプト、dbt、Python スクリプト
クエリエンジン	DuckDB / MotherDuck	—
DataFrame	polars	—
ノートブック	marimo	—
プロジェクト管理	uv	—
可視化	altair, seaborn	—

言語の優先順位

まず SQL（DuckDB 方言）、次に Python、最後に bash を使います。目的を達成できる最もシンプルな言語を選んでください。

他のスキルへの相互参照

このスキルでは、取り込み／変換／ノートブック作業向けの独自のスタックを扱います。基盤となるツールについては、次を参照してください。

• DuckDB の構文（Friendly SQL、ファイルの読み込み、ネスト型）→ motherduck-duckdb-sql または duckdb-docs
• MotherDuck の接続、探索、共有、Dives → motherduck-* スキル
• marimo ノートブック（セル、リアクティビティ、SQL セル）→ marimo-notebook
• ローカル DuckDB セッション管理 → attach-db、query、read-file

プロジェクト構成

データプロジェクトでは、次の構成に従ってください。

project/
├── ingest/              # Extraction and loading scripts
│   ├── <source>.py      # One file per data source
│   └── .dlt/            # dlt config (if using dlt)
├── transform/           # Transformation logic
│   ├── models/          # SQL models (sqlmesh or plain SQL)
│   └── config.yaml      # sqlmesh config (if using sqlmesh)
├── notebooks/           # Exploration and analysis
│   └── *.py             # marimo notebooks (plain .py files)
├── data/                # Local data files (gitignored)
├── pyproject.toml       # Dependencies
├── uv.lock              # Locked dependencies (committed)
└── *.duckdb             # Local database (gitignored)

すべてのディレクトリが必要なわけではありません。シンプルな分析なら、notebooks/ と DuckDB ファイルだけで済むこともあります。必要に応じて拡張してください。

uv — プロジェクト管理

pip を直接使わないでください。Python に関する作業はすべて uv を通して行います。

uv init my-project                    # New project
uv add polars duckdb                  # Add dependencies
uv sync                               # Install into .venv
uv run python script.py               # Run in project venv
uv run --with requests script.py      # Ad-hoc dependency

スタンドアロンスクリプトでは、インラインスクリプト依存関係（PEP 723）を使います。

# /// script
# dependencies = ["dlt[duckdb]", "polars"]
# requires-python = ">=3.12"
# ///

uv run script.py で実行すると、依存関係が自動的に解決されます。

uv.lock は必ずコミットしてください。依存関係の宣言には pyproject.toml を使い、requirements.txt は使わないでください。

polars — DataFrame

複雑な文字列変換、ML 機能、行単位の条件分岐など、Python のロジックが必要な場合は polars を使います。結合、集計、ウィンドウ関数には SQL を優先してください。

主なパターン

import polars as pl

# Lazy evaluation (always prefer for production)
lf = pl.scan_parquet("events/*.parquet")
result = (
    lf.filter(pl.col("event_date") >= "2024-01-01")
    .group_by("user_id")
    .agg(pl.col("amount").sum().alias("total_spend"))
    .sort("total_spend", descending=True)
    .collect()
)

# Three contexts
df.select(...)         # Pick/transform columns (output has ONLY these)
df.with_columns(...)   # Add/overwrite columns (keeps all originals)
df.filter(...)         # Keep rows matching condition

DuckDB との相互運用（Arrow によるゼロコピー）：

import duckdb
result = duckdb.sql("SELECT * FROM df WHERE amount > 100").pl()

dlt — 取り込み

プロジェクトで取り込みに dlt を使う場合です。API 呼び出し、ページネーション、スキーマ推論、増分読み込み、状態管理を処理します。

スキャフォールドと実行

dlt init rest_api duckdb             # Scaffold pipeline
uv run python pipeline.py            # Run extraction
dlt pipeline <name> info             # Inspect state
dlt pipeline <name> schema           # View inferred schema

パイプラインのパターン

最小限のパイプライン：

import dlt

pipeline = dlt.pipeline(
    pipeline_name="my_pipeline",
    destination="duckdb",
    dataset_name="raw",
)
info = pipeline.run(data, table_name="events")

増分読み込み：

@dlt.resource(write_disposition="merge", primary_key="id")
def users(updated_at=dlt.sources.incremental("updated_at")):
    yield from fetch_users(since=updated_at.last_value)

REST API ソース（宣言型）：

from dlt.sources.rest_api import rest_api_source

source = rest_api_source({
    "client": {"base_url": "https://api.example.com/v1"},
    "resource_defaults": {"primary_key": "id", "write_disposition": "merge"},
    "resources": [
        "users",
        {
            "name": "events",
            "write_disposition": "append",
            "endpoint": {
                "path": "events",
                "incremental": {"cursor_path": "created_at", "initial_value": "2024-01-01"},
            },
        },
    ],
})

書き込み方式：

書き込み方式	動作	用途
append	行を挿入（デフォルト）	不変のイベント、ログ
replace	削除して再作成	小規模なルックアップテーブル
merge	primary_key でアップサート	可変レコード

宛先： duckdb（ローカルファイル）、motherduck（クラウド）。motherduck_token 環境変数を設定するか、.dlt/secrets.toml で構成してください。

sqlmesh — 変換

プロジェクトで変換に sqlmesh を使う場合です。SQL を優先し、plan/apply ワークフローで、誤って本番環境を変更することを防ぎます。

スキャフォールドと実行

sqlmesh init duckdb                              # New project
sqlmesh init -t dlt --dlt-pipeline <name>        # From dlt schema
sqlmesh plan                                     # Preview + apply (dev)
sqlmesh plan prod                                # Promote to production
sqlmesh fetchdf "SELECT * FROM analytics.users"  # Ad-hoc query
sqlmesh test                                     # Run unit tests
sqlmesh ui                                       # Web interface

モデルの種類

種類	動作	用途
FULL	テーブル全体を書き換えます	小規模なディメンションテーブル
INCREMENTAL_BY_TIME_RANGE	新しい期間を処理します	ファクト、イベント、ログ
INCREMENTAL_BY_UNIQUE_KEY	キーでアップサートします	変更可能なディメンション
SEED	静的な CSV データ	参照／ルックアップデータ
VIEW	SQL ビュー	単純なパススルー
SCD_TYPE_2	緩やかに変化するディメンション	履歴の追跡

モデルの例

MODEL (
    name analytics.stg_events,
    kind INCREMENTAL_BY_TIME_RANGE (time_column event_date),
    cron '@daily',
    grain (event_id),
    audits (NOT_NULL(columns=[event_id]))
);

SELECT
    event_id,
    user_id,
    event_type,
    event_date
FROM raw.events
WHERE event_date BETWEEN @start_date AND @end_date

設定（config.yaml）

gateways:
  local:
    connection:
      type: duckdb
      database: db.duckdb
default_gateway: local
model_defaults:
  dialect: duckdb

dlt との統合

sqlmesh init -t dlt は、dlt が推測したスキーマから外部モデルと増分ステージングモデルを自動生成します。dlt によるスキーマの変更は、sqlmesh plan で検出されます。