ストーリー

ETL vs ELT

// ETL: Extract → Transform → Load
// 変換してからデータウェアハウスにロード
class ETLPipeline {
  async run(): Promise<void> {
    // 1. Extract: ソースからデータを抽出
    const rawData = await this.extractFromSources();

    // 2. Transform: データを変換・クレンジング
    const transformed = this.transform(rawData);
    // 型変換、欠損値処理、集約、デノーマライズ

    // 3. Load: データウェアハウスに格納
    await this.loadToWarehouse(transformed);
  }
}

// ELT: Extract → Load → Transform
// まずロードしてからウェアハウス内で変換（モダンなアプローチ）
class ELTPipeline {
  async run(): Promise<void> {
    // 1. Extract: ソースからデータを抽出
    const rawData = await this.extractFromSources();

    // 2. Load: データレイクにそのまま格納
    await this.loadToDataLake(rawData);

    // 3. Transform: データレイク/ウェアハウス内で変換
    // dbt等のツールでSQLベースの変換
    await this.transformInWarehouse();
  }
}

const ETL_VS_ELT = {
  etl: {
    bestFor: "構造化データ、明確なスキーマ",
    tools: ["Apache Airflow", "AWS Glue", "Informatica"],
    pros: "データ品質を事前に保証",
    cons: "スキーマ変更に弱い、変換がボトルネック",
  },
  elt: {
    bestFor: "大量データ、スキーマが頻繁に変わる",
    tools: ["dbt", "BigQuery", "Snowflake", "Databricks"],
    pros: "柔軟、スケーラブル、変換を後から追加",
    cons: "ストレージコスト増、データレイクの管理が必要",
  },
};

バッチ処理 vs ストリーム処理

// バッチ処理: 定期的にまとめて処理
interface BatchProcessing {
  schedule: "1時間ごと / 1日ごと";
  latency: "分〜時間単位の遅延";
  tools: ["Apache Spark", "AWS EMR", "Hadoop"];
  useCase: "日次レポート、月次集計、モデル学習";
}

// ストリーム処理: リアルタイムでデータを処理
interface StreamProcessing {
  schedule: "常時（イベント駆動）";
  latency: "ミリ秒〜秒単位";
  tools: ["Apache Kafka", "Apache Flink", "Kinesis Data Streams"];
  useCase: "リアルタイムダッシュボード、異常検知、不正検知";
}

// Lambda Architecture: バッチ + ストリームの組み合わせ
class LambdaArchitecture {
  // バッチレイヤー: 全データを定期的にバッチ処理
  batchLayer = {
    input: "データレイク全体",
    output: "バッチビュー（正確だが遅い）",
    schedule: "毎日or毎時",
  };

  // スピードレイヤー: 新しいデータをリアルタイムで処理
  speedLayer = {
    input: "リアルタイムストリーム",
    output: "リアルタイムビュー（速いが近似）",
    schedule: "常時",
  };

  // サービングレイヤー: バッチビュー + リアルタイムビューをマージ
  servingLayer = {
    query: "バッチビュー + リアルタイムビュー",
    result: "最新かつ正確なデータ",
  };
}

ハイレベル設計

┌────────────────────────────────────────────────────────┐
│                                                        │
│  データソース群                                          │
│  [RDB] [API] [ログ] [IoT] [ファイル]                     │
│     │     │    │      │       │                        │
│     └──────┴────┴──────┴───────┘                        │
│                │                                        │
│         [データ収集層]                                    │
│    Kafka / Kinesis / Fluentd                            │
│                │                                        │
│         ┌──────┴──────┐                                 │
│         ▼             ▼                                 │
│    [バッチ処理]    [ストリーム処理]                        │
│    Spark / EMR     Flink / Kinesis                      │
│         │             │                                 │
│         ▼             ▼                                 │
│    [データレイク]  [リアルタイムDB]                        │
│    S3 / GCS       Redis / DynamoDB                      │
│         │                                               │
│         ▼                                               │
│    [データウェアハウス]                                    │
│    BigQuery / Redshift / Snowflake                      │
│         │                                               │
│         ▼                                               │
│    [BIツール / ダッシュボード]                              │
│    Metabase / Grafana / Tableau                         │
│                                                        │
└────────────────────────────────────────────────────────┘

データ品質の担保

// データ品質チェックの実装
class DataQualityChecker {
  checks = [
    {
      name: "完全性チェック",
      description: "NULLや欠損値の検出",
      rule: (data: Row[]) => data.filter(r => r.userId == null).length === 0,
    },
    {
      name: "一意性チェック",
      description: "重複レコードの検出",
      rule: (data: Row[]) => new Set(data.map(r => r.id)).size === data.length,
    },
    {
      name: "鮮度チェック",
      description: "データの最新性を確認",
      rule: (data: Row[]) => {
        const latestTimestamp = Math.max(...data.map(r => r.timestamp));
        return Date.now() - latestTimestamp < 3600000; // 1時間以内
      },
    },
    {
      name: "整合性チェック",
      description: "参照整合性やビジネスルールの検証",
      rule: (data: Row[]) => data.every(r => r.amount >= 0),
    },
  ];

  async validate(data: Row[]): Promise<QualityReport> {
    const results = this.checks.map(check => ({
      name: check.name,
      passed: check.rule(data),
    }));

    const allPassed = results.every(r => r.passed);
    if (!allPassed) {
      await this.alertService.notify("データ品質アラート", results);
    }

    return { results, allPassed };
  }
}

まとめ

チェックリスト

• ETLとELTの違いを説明できた
• バッチ処理とストリーム処理の使い分けを理解した
• Lambda Architectureの仕組みを把握した
• データ品質チェックの重要性を理解した

次のステップへ

次は「分析基盤の設計」を学びます。データウェアハウスとOLAP、ダッシュボードの設計を掘り下げます。

推定読了時間: 30分