dltを用いてSnowflakeに格納したデータをdbtで整形してみた

初めに

dbtとは

前準備

データの格納

プロジェクトの作成

pip install dbt-core dbt-snowflake

dbt init dbt_weather

weather_pipeline/            # プロジェクトルートディレクトリ
├── dlt_pipeline/            # dlt関連ファイル (ここで dlt init 実行)
│   ├── .dlt/                # secrets.toml等を格納するディレクトリ。ここから認証情報を読み取る
│   ├── dlt_weather_test.py  # APIを実行しSnowflakeにデータを格納するコード
│   └── ...
├── dbt_weather/             # dbt関連ファイル (ここで dbt init 実行)
│   ├── dbt_project.yml.     # dbtにプロジェクトの操作方法を指示するファイル
│   ├── models/              # 整形するSQLクエリを格納するディレクトリ
│   └── ...

profiles.yamlの作成

dbt_weather:  # 任意のプロファイル名（dbt_project.yml の profile: に一致させる）
  target: dev
  outputs:
    dev:
      type: snowflake  # 固定（Snowflakeの場合）
      account: <account_id>  # SnowflakeのアカウントID（URLの最初の部分）
      user: <user_name>  # Snowflake のユーザー名（例: JOHN_SMITH）

      role: <role_name>  # 使用するロール（権限レベルの高いもの推奨。例: SYSADMIN）

      # キーペア認証設定
      private_key_path: <key_path>  # 秘密鍵ファイルのフルパス（通常は ~/.ssh 配下など）
      # または下記のように文字列を直接指定も可能（非推奨・セキュリティ上注意）:
      # private_key: "-----BEGIN PRIVATE KEY-----\nMIIEvQIBADANBgkqh..."

      private_key_passphrase: <key_passphrase>  # 鍵作成時に設定したパスフレーズ。未設定なら省略可能

      database: DLT_WEATHER  # 使用するデータベース名（例: ANALYTICS）
      warehouse: COMPUTE_WH  # 使用するウェアハウス（例: COMPUTE_WH）
      schema: DBT_WEATHER_TEST  # 使用するスキーマ（例: PUBLIC）

      threads: 1  # 並列実行数。1〜4程度で始めればOK
      client_session_keep_alive: False  # 接続の維持（長時間処理がある場合 True にすることも）

      query_tag: dbt_weather_run  # 任意のクエリ識別タグ（監査・ログ用途）

      # ⏱️ 以下はオプション（必要に応じて設定）
      connect_retries: 0  # 接続リトライ回数
      connect_timeout: 10  # タイムアウト秒数
      retry_on_database_errors: False  # データベースエラーでの自動再試行を有効にするか
      retry_all: False  # すべての失敗で再試行を行うか
      reuse_connections: True  # 接続の再利用を行うか

モデルの作成

//WordPressのエラー解消のために全文コメントアウトしています。

/*
WITH source_data AS (
    SELECT
        MAIN__TEMP_MAX AS max_temp,
        MAIN__TEMP_MIN AS min_temp,
        name AS prefectures
    FROM {{ source('DBT_WEATHER_TEST', 'weather') }}
    WHERE name IN ('Tokyo', 'Aichi', 'Fukuoka')
)

SELECT *
FROM source_data
*/

version: 2

sources:
  - name: DBT_WEATHER_TEST           # dbt上でのschema名（任意の別名ではなく、Snowflake上のスキーマ名に合わせる）
    database: DLT_WEATHER     # ← DLTが書き込んだSnowflakeのDB名（例: WEATHER_DB）
    schema: DBT_WEATHER_TEST
    tables:
      - name: weather

models:
  dbt_weather:
    # Config indicated by + and applies to all files under models/example/
      +materialized: view

データの取得・格納・変換

import dlt
import pandas as pd
from dlt.sources.helpers import requests
import subprocess

API_key = dlt.secrets["openweathermap"]["API_key"]

df = pd.read_excel("prefecture.xlsx")  # A列にidがある
city_list = df["City_id"].tolist()  # "City_id"列をリスト化


weather_data = []


# 各都道府県についてAPIを実行
for city in city_list:
    url = f"https://api.openweathermap.org/data/2.5/weather?id={city}&appid={API_key}"
    response = requests.get(url)

    if response.status_code == 200:
        weather_data.append(response.json())  # JSONデータをリストに追加

    else:
        print(f"Error fetching data for {city}: {response.status_code}")

weather_df = pd.DataFrame(weather_data)

pipeline = dlt.pipeline(
    pipeline_name="weather_snowflake",
    destination="snowflake",
    dataset_name="DBT_WEATHER_TEST",
    
)

#実際のコードでは@を_に変えてください
load_info = pipeline.run(weather_data, table@name="weather")

try:
    subprocess.run(
        ["dbt", "run", "--project-dir", "../dbt_weather"],
        check=True
    )
    print("dbt run completed successfully.")
except subprocess.CalledProcessError as e:
    print("dbt run failed:", e)

実行結果

整形したテーブル

データリネージ

dbt docs generate
dbt docs serve

最後に

参考

• dbtのCLI版をインストールして使ってみた
• dbtで始めるデータパイプライン構築〜入門から実践〜
• Snowflake setup
• DuckDBにdltでロードしたデータをdbtで整形してevidenceで可視化した

この記事の著者：大槻南央