wren-dlt-connector

wren-dlt-connector

SaaS データを Wren Engine に接続して SQL 分析を行います。1 回の会話で、ゼロから検証済みのクエリ可能なプロジェクトを作成します。

対象ユーザー

SQL とある程度の Python の知識があり、dlt や Wren は初めて使う可能性があるデータアナリストです。概念が初めて登場する際は簡潔に説明しますが、SQL を理解している人がすでに知っている内容を過度に説明しないでください。

概要

このスキルでは、次の 4 段階のワークフローを進めます。

1. 抽出 — dlt（data load tool）を使って、SaaS API からローカルの DuckDB ファイルにデータを取り込みます
2. モデリング — DuckDB スキーマを調査し、Wren のセマンティックプロジェクト（YAML モデル、リレーションシップ、プロファイル）を自動生成します
3. 構築と検証 — プロジェクトを構築し、実際に SQL クエリを実行して、すべてがエンドツーエンドで動作することを確認します
4. 引き継ぎ — ユーザーにデータと次のステップを示します

ユーザーはどの段階からでも開始できます。どの段階から始めるのかを尋ねてください。すでに .duckdb ファイルがあり、フェーズ 2〜4 だけが必要な場合もあります。

目標は、正しそうに見えるファイルを作ることではなく、実際にクエリが成功するプロジェクトを作ることです。 成功を宣言する前に、必ず検証ステップを実行してください。

重要：DuckDB のカタログ名

wren engine が DuckDB ファイルに接続すると、ファイル名から .duckdb 拡張子を除いた名前をカタログのエイリアスとして使用して ATTACH します。

ATTACH DATABASE 'stripe_data.duckdb' AS "stripe_data" (READ_ONLY)

つまり、すべてのモデルの table_reference.catalog は、DuckDB ファイル名の拡張子を除いた部分と一致している必要があります。ファイルが hubspot.duckdb なら、カタログは hubspot です。ファイルが my_pipeline.duckdb なら、カタログは my_pipeline です。

ここを間違えると、クエリ実行時に「table not found」エラーが発生します。introspect_dlt.py スクリプトがこれを自動的に処理します。

重要：型の正規化

列の型は、wren SDK の type_mapping.parse_type() 関数を使って正規化する必要があります。この関数は sqlglot を使い、DuckDB の HUGEINT や TIMESTAMP WITH TIME ZONE などのデータベース固有の型を、wren-core が理解できる標準 SQL 型に変換します。型の対応関係をハードコードしないでください。必ず parse_type(raw_type, "duckdb") に委ねてください。

wren SDK がインストールされている場合、introspect_dlt.py スクリプトがこれを自動的に処理します。

フェーズ 1：抽出 — dlt パイプラインのセットアップ

ステップ 1：SaaS のデータソースを選ぶ

接続する SaaS サービスをユーザーに尋ねてください。一般的で検証済みのデータソースと認証要件の一覧については、references/dlt_sources.md を参照してください。データソースが一覧にない場合は、dlthub.com/docs/dlt-ecosystem/verified-sources を検索して、dlt に検証済みのデータソースがあるか確認してください。

ステップ 2：dlt をインストールする

pip install "dlt[duckdb]" --break-system-packages

ステップ 3：パイプラインスクリプトを作成する

次の処理を行う Python スクリプトを作成します。

1. 選択した SaaS 用の dlt ソース関数をインポートする
2. destination='duckdb' とローカルファイルパスを指定してパイプラインを設定する
3. pipeline.run(source) でパイプラインを実行する

基本的なパターンは次のとおりです。ソースに合わせて調整してください（ソース別のテンプレートは references/dlt_sources.md を参照してください）。

import dlt

pipeline = dlt.pipeline(
    pipeline_name="<source>_pipeline",
    destination="duckdb",
    dataset_name="<source>_data",
)

# Source-specific: check references/dlt_sources.md for auth patterns
source = <source_function>(api_key=dlt.secrets.value)

info = pipeline.run(source)
print(info)

ステップ 4：認証情報を設定する

dlt は環境変数または .dlt/secrets.toml から認証情報を読み込みます。1 回だけ実行する場合は、次の方法が最も簡単です。

# Set the credential as an environment variable
# The exact variable name depends on the source — check references/dlt_sources.md
export SOURCES__<SOURCE>__API_KEY="the-actual-key"

ユーザーに API キーまたはトークンを尋ねてください。次の点を伝えてください。

• 認証情報を git にコミットしないでください
• 1 回だけ実行する場合は、環境変数を使うのが最も簡単です
• 繰り返し使用する場合は、.dlt/secrets.toml を作成できます

ステップ 5：パイプラインを実行する

python <pipeline_script>.py

実行後、次の点を確認してください。

1. パイプラインがエラーなく完了したこと
2. .duckdb ファイルが作成されたこと（通常は <pipeline_name>.duckdb）
3. 検出されたテーブルと各テーブルの列数を表示すること

import duckdb
con = duckdb.connect("<pipeline_name>.duckdb", read_only=True)
for row in con.execute("""
    SELECT table_schema, table_name,
           (SELECT COUNT(*) FROM information_schema.columns c
            WHERE c.table_schema = t.table_schema AND c.table_name = t.table_name) as col_count
    FROM information_schema.tables t
    WHERE table_schema NOT IN ('information_schema', 'pg_catalog')
      AND table_name NOT LIKE '_dlt_%'
    ORDER BY table_schema, table_name
""").fetchall():
    print(f"  {row[0]}.{row[1]} ({row[2]} columns)")
con.close()

Phase 2: モデル — Wren プロジェクトを生成する

イントロスペクションスクリプトを実行して、DuckDB ファイルから完全な Wren プロジェクトを自動生成します。

python <path-to-this-skill>/scripts/introspect_dlt.py \
    --duckdb-path <path-to-duckdb-file> \
    --output-dir <project-directory> \
    --project-name <name>

このスクリプトは、次の処理を行います。

• DuckDB ファイルに読み取り専用で接続します
• table_reference.catalog を DuckDB のファイル名の拡張子を除いた部分に設定します（wren エンジンの ATTACH の動作に合わせます）
• information_schema を使用して、すべてのテーブルとカラムを検出します
• dlt の内部テーブル（_dlt_loads、_dlt_pipeline_state など）を除外します
• モデル定義から dlt のメタデータカラム（_dlt_id、_dlt_load_id、_dlt_list_idx）を除外します
• _dlt_parent_id カラムとテーブルの命名規則から、親子関係を検出します
• wren.type_mapping.parse_type()（sqlglot ベース）を使用してカラム型を正規化します
• 完全な v2 YAML プロジェクト（wren_project.yml、models/、relationships.yml、instructions.md）を生成します

実行後、生成された内容をユーザーに表示します。

# Show project summary
cat <project-directory>/wren_project.yml
echo "---"
ls <project-directory>/models/
echo "---"
cat <project-directory>/relationships.yml

モデルの正確性を確認する

生成されたモデルを 1 つ抜き取り、次の点を確認します。

1. table_reference.catalog が DuckDB のファイル名と一致している（例：stripe_data.duckdb の場合は stripe_data）
2. table_reference.schema が DuckDB のスキーマと一致している（通常は main）
3. カラム一覧に _dlt_* カラムがない
4. カラム型が妥当である（VARCHAR、BIGINT、BOOLEAN、TIMESTAMP など）

接続プロファイルを設定する

接続情報を毎回指定しなくてもユーザーがクエリを実行できるように、Wren プロファイルを作成します。url は .duckdb ファイル自体ではなく、そのファイルを含むディレクトリを指定する必要があります。

import yaml
from pathlib import Path

wren_home = Path.home() / ".wren"
wren_home.mkdir(exist_ok=True)
profiles_file = wren_home / "profiles.yml"

existing = (yaml.safe_load(profiles_file.read_text()) or {}) if profiles_file.exists() else {}
existing.setdefault("profiles", {})

profile_name = "<source>_dlt"
existing["profiles"][profile_name] = {
    "datasource": "duckdb",
    "url": str(Path("<duckdb-path>").resolve().parent),
    "format": "duckdb",
}
existing["active"] = profile_name

profiles_file.write_text(yaml.dump(existing, default_flow_style=False, sort_keys=False))

Phase 3: ビルドと検証 — プロジェクトを実際に動作させる

このフェーズは省略できません。YAML を生成できても、クエリ時に失敗するプロジェクトは成功とはいえません。

Step 1: MDL をビルドする

cd <project-directory>
wren context build

これにより、YAML モデルが target/mdl.json にコンパイルされます。失敗した場合は、先に問題を修正してから次に進みます（「トラブルシューティング」を参照）。

Step 2: 実際のクエリで検証する

プロジェクトが機能することを確認するため、生成されたモデルごとに少なくとも 1 つのクエリを実行します。

# For each model, verify it resolves correctly
wren --sql 'SELECT COUNT(*) as total FROM "<table_name>"'

クエリが失敗した場合は、先に進む前にモデルをデバッグして修正します。よくある問題は次のとおりです。

• table_reference のカタログが間違っている → 「table not found」
• 型が一致しない → metadata.yml でカラム型を修正します
• プロファイルがない → wren profile list を確認します

Step 3: 興味深いクエリを実行する

基本的なクエリが成功したら、さらに 2～3 件の興味深いクエリを実行し、データの内容をユーザーに示します。

# Preview data
wren --sql 'SELECT * FROM "<table_name>" LIMIT 5'

# If there's a relationship, verify both models are queryable
wren --sql 'SELECT * FROM "<parent>" LIMIT 5'
wren --sql 'SELECT * FROM "<child>" LIMIT 5'

結果をユーザーに示し、何が表示されているのかを説明します。これは Wren を通してデータを見る最初の機会です。印象に残るものにしましょう。

Step 4: 成功を確認する

クエリから実際のデータが返ってきた後に限り、セットアップが完了したことをユーザーに伝えます。次の内容を要約します。

• 作成したモデルの数
• 検出されたリレーションシップ
• アクティブなプロファイル
• 次に試せるクエリの例

トラブルシューティング

wren context build が失敗する場合：

• wren_project.yml で data_source: duckdb が設定されていることを確認します
• プロファイル内の DuckDB ファイルパスが正しいことを確認します
• 詳細なエラーメッセージを確認するには、wren context validate を実行します

クエリが「table not found」で失敗する場合：

• 最も可能性の高い原因： table_reference.catalog が DuckDB のファイル名と一致していません。ファイル名が pipeline.duckdb の場合、カタログは空文字列ではなく pipeline にする必要があります。
• プロファイルの url が .duckdb ファイルを含むディレクトリを指していることを確認します
• アンダースコアが 2 つ続くテーブル名は引用符で囲む必要があります："hubspot__contacts"

クエリが型エラーで失敗する場合：

• モデル YAML のカラム型を確認します。正規の SQL 型（VARCHAR、BIGINT など）を指定する必要があります
• 適切に型を正規化するには、wren SDK をインストールした状態で introspect_dlt.py を再実行します

一般：

• プロファイルがアクティブであることを確認します：wren profile list
• dlt パイプラインの実行中は DuckDB ファイルがロックされている可能性があります。完了するまで待ちます

重要な注意事項

• dlt の _dlt_parent_id／_dlt_id 列は実際の DuckDB テーブルには保持されますが、Wren のモデル定義では非表示になります。これらはリレーション条件でのみ使用されます。
• DuckDB には単一ライターの制限があります。クエリの実行中に dlt の同期を実行しないでください。並行アクセスを行う場合、dlt は別のファイルに書き込み、アトミックに切り替える必要があります。
• 生成されるモデルでは、dlt が作成した DuckDB テーブルに直接対応するため、ref_sql ではなく table_reference を使用します。
• 列の型は、sqlglot の DuckDB 方言を使用する wren SDK の parse_type() で正規化されます。型に誤りがある場合は、ユーザーがモデルの metadata.yml を直接編集できます。