はじめに
気象庁のWebサイトを眺めていたところ、
全国26地点の沿岸代表地点における波浪推算値データをcsv形式でダウンロードできることに気が付きました。
ということで、Snowflake,dbt,Streamlitのお勉強をしようと思い
Snowflake + dbt + Streamlit を使って時系列分析ダッシュボードを作ってみる
をやってみました。🌊
データ概要
- 観測地点: 全国26地点
- 観測頻度: 1日2回（0時、12時）
-
データ項目:
- 周期(秒)
- 波高(m)
- 波向(16方位)
- 月次単位でCSVダウンロード可能
2025年6月データの一部
地点 日 時 周期(s) 波高(m) 波向(16方位)
A 1 0 6 0.6 3
A 1 12 5 0.6 3
A 2 0 6 0.7 3
A 2 12 7 0.5 3
使用技術スタック
| 技術 | 用途 |
|---|---|
| Snowflake | クラウドデータウェアハウス（データ保存） |
| dbt | データ変換パイプライン |
| Streamlit | ダッシュボードUI（可視化） |
| Python | データアップロード自動化 |
→ Snowflake,dbtは少し触ったことがある
→ Streamlitは、ほとんど経験無し。。
作成物の全体像（アーキテクチャ）
スキーマ設計
| スキーマ | 役割 |
|---|---|
| source | 生CSVデータ（月次テーブル） |
| raw | 全月次テーブルの自動UNION |
| cleansing | カラム名英語化 + 日時統合 |
| base | 地点マスタJOIN（地点名、緯度経度追加） |
Snowflake初期設定
データ格納用に以下を作成しました。
create or replace database wave_db;
create or replace schema wave_db.source;
create or replace stage wave_db.source.int_stage;
CSVアップロード自動化（Python）
月次CSVを 手動作業なしでSnowflakeにロード するため、
Pythonスクリプトを作成しました。
主な機能:
- 文字コード自動判定（UTF-8/Shift_JIS）
- CSVヘッダからカラム名自動取得
- テーブル自動作成
- 内部ステージ経由でCOPY
実行ログ（抜粋）:
11:37:13 [INFO] [2/2] gmdf1812.csv → GMDF1812
11:37:13 [INFO] ======================================================================
11:37:13 [INFO] 📤 アップロード開始: gmdf1812.csv → GMDF1812
11:37:13 [INFO] ======================================================================
11:37:13 [WARNING] ⚠️ CSVファイルは UTF-8 ではなく shift_jis で読み込まれました
11:37:13 [INFO] ✅ CSVカラム取得完了 (encoding=shift_jis): ['地点', '日', '時', '周期(s)', '波高(m)', '波向(16方位)']
11:37:13 [INFO] Snowflakeに接続中... (account={sf_config.account}, user={sf_config.user})
11:37:14 [INFO] ✅ 認証成功
11:37:14 [INFO] ✅ コンテキスト設定完了 (role=accountadmin, warehouse=COMPUTE_WH, database=wave_db, schema=source)
11:37:14 [INFO] テーブル作成中: GMDF1812 (6 カラム, 型=VARCHAR(100))
11:37:14 [INFO] SQL: CREATE OR REPLACE TABLE GMDF1812 ("地点" VARCHAR(100), "日" VARCHAR(100), "時" VARCHAR(100), "周期(s)" VARCHAR(100), "波高(m)" VARCHAR(100), "波向(16方位)" VARCHAR(100))
11:37:14 [INFO] ✅ テーブル作成完了: GMDF1812
11:37:14 [INFO] ✅ ステージ確認完了: int_stage
11:37:14 [INFO] ✅ ファイルアップロード完了: gmdf1812.csv → @int_stage
11:37:14 [INFO] データコピー中: @int_stage/gmdf1812.csv → GMDF1812
11:37:14 [INFO] SQL: COPY INTO GMDF1812 FROM @int_stage/gmdf1812.csv FILE_FORMAT = (TYPE = 'CSV' SKIP_HEADER = 1)
11:37:15 [INFO] ✅ データコピー完了: 1612 行
11:37:15 [INFO] Snowflakeセッションを終了しました
11:37:15 [INFO] ======================================================================
11:37:15 [INFO] 🎉 アップロード完了: GMDF1812
11:37:15 [INFO] ======================================================================
11:37:15 [INFO]
11:37:15 [INFO] ======================================================================
11:37:15 [INFO] 📊 フォルダ一括アップロード 結果サマリー
11:37:15 [INFO] ======================================================================
11:37:15 [INFO] ✅ 成功: 2 ファイル
11:37:15 [INFO] ======================================================================
dbtによるデータパイプライン
工夫ポイント:動的テーブルUNION
毎月CSVが追加されるたびに raw モデルを書き換えるのは大変なので、
dbtマクロで source スキーマ内のテーブルを自動取得 するようにしました。
{% macro union_all_source_tables() %}
{% set tables = run_query("SHOW TABLES IN wave_db.source") %}
{% for table in tables %}
SELECT *, '{{ table.name }}' as source_table
FROM {{ source('wave_db', table.name) }}
{% if not loop.last %} UNION ALL {% endif %}
{% endfor %}
{% endmacro %}
工夫ポイント:地点マスタ管理
地点コード（A〜Z）と地点名・緯度経度を
dbt seeds でマスタ管理 し、base層でJOINしています。
location_code,location_name,latitude,longitude
A,網走沖,44.0,144.25
B,釧路沖,42.92,145.08
Streamlit ダッシュボード
Snowflake接続成功 & データ概要表示
Snowflakeに接続し、データ件数や期間を表示できているーー！！！
特定地点の波高時系列グラフ（30日間）
地点を選択すると、波高の時系列推移を可視化 できたー！
まとめ
- Snowflake + dbt + Streamlit の組み合わせは学習コストも低めでした！
- dbt初心者・Streamlit初心者でも分析 → 可視化まで一気通貫で作れる！