ETL: FlyData(ELT)とXplenty(ETL)を組み合わせた循環型データ統合アーキテクチャ

​

FlyDataについて

​

FlyDataの特徴

• CDCによるリアルタイム連携: DB側の変更をほぼ即座に検知して転送
• セットアップが軽量: 最小限の設定でレプリケーションを開始できる
• DWH連携に特化: データウェアハウスへの取り込みに最適化されたアーキテクチャ

​

概要：なぜ「循環型」なのか

​

アーキテクチャ全体像

​

各ステップの役割

• ステップ1：FlyDataによる収集（ELT）
  • CDCにより各運用DBの変更をリアルタイム〜準リアルタイムでSnowflakeに連携
  • 変換は行わず、まず生データをそのままDWHへ保存する（ELTの考え方の核）
  • 異なる種類の複数DBを1つのDWHに統合できる
• ステップ2：Snowflake内での分析・計算
  • 蓄積した生データをもとに、View・ストアドプロシージャ・dbtなどで調整済みデータを作成
  • 例：在庫の調整数量、精算金額、集計済みのユーザースコアなど
• ステップ3：Xplentyによる書き戻し（ETL）
  • XplentyのビジュアルパイプラインでSnowflake上の加工データを読み込み
  • 必要な変換（フィールドマッピング、型変換、フィルタ）を適用
  • Merge（Upsert）、Append、Truncate & Insertなどの方式で各運用DBへ書き戻す

​

ユースケース：ECプラットフォームの決済・在庫データ同期

​

背景

• 日本・米国・韓国それぞれに個別の運用DBを持つECサービス
• 各リージョンの決済データを中央で集計し、グローバルな決済調整額を算出したい
• 算出した調整額を各リージョンのDBへ反映する必要がある

​

ステップ1：FlyDataで各リージョンDB → Snowflakeへ収集

• 日本のMySQL → CDC → Snowflake: raw.jp_payments
• 米国のPostgreSQL → CDC → Snowflake: raw.us_payments
• 韓国のMySQL → CDC → Snowflake: raw.kr_payments

​

ステップ2：Snowflakeでグローバル調整額を算出

-- 例：全体平均を基準にリージョンごとの調整額を算出
CREATE OR REPLACE VIEW analytics.adjusted_payments AS
SELECT
   payment_id,
   'JP' AS region,
   jp.amount - (global_avg.avg_amount * 0.3) AS adjusted_amount
...
...;
-- US、KRについても同様のロジックを適用

​

ステップ3：Xplentyで各リージョンDBへ書き戻し

[Snowflake Source]
 analytics.adjusted_payments
       ↓
[Filterコンポーネント]
 region = 'JP' で絞り込み
       ↓
[Selectコンポーネント]
 payment_id, adjusted_qty のフィールドマッピング・型変換
       ↓
[Database Destination: JP MySQL]
 テーブル: payment_adjustment
 モード: Merge（Upsert）by payment_id

• JP MySQL、US PostgreSQL、KR MySQLそれぞれに個別のパイプラインを用意するか、単一パイプライン内で分岐させる
• SnowflakeでのView更新完了後にXplentyパイプラインが自動実行されるよう、依存関係をスケジューラ側で設定する

​

メリット・デメリット

​

メリット

​

デメリット

​

運用時の注意点

1. データループの防止（最重要）
   • 書き戻すデータにはsource_systemやis_adjusted、updated_byのような出所を示すメタカラムを付与する
   • FlyDataのCDCフィルタ、またはSnowflake側のViewロジックで、書き戻したデータを再収集対象から除外する
2. 冪等性の担保
   • Xplentyの書き戻しパイプラインが再実行されても重複挿入が起きないよう、Merge（Upsert）モードと明確な主キーを設定する
3. 実行順序の管理
   • Snowflake側の計算（View更新、dbt runなど）が完了してからXplentyパイプラインが動くよう、スケジュールや依存関係トリガーを設定する
   • Xplentyの Dependent Execution や Webhook Trigger の活用が有効
4. 失敗時の部分ロールバック対策
   • 書き戻しの途中で一部のDBだけ更新され残りが失敗すると、データ不整合が生じる
   • Xplentyの Single Transaction Mode や Pre/Post-action SQL を使ってロールバック戦略を用意する
5. 監視とアラート
   • FlyData・Xplenty双方に障害通知を設定し、どの区間で失敗してもすぐ気づけるようにする
   • Slack、PagerDuty、Emailなどのフックが有効
6. 機密データの取り扱い
   • 個人情報や金融情報を含むデータを書き戻す場合は、XplentyのSelectやFilterコンポーネントで不要な機密項目を除外・マスキングする

​

まとめ