Hadoop アーキテクチャを見直す — HDFS、YARN、そしてその先へ

┌──────────────────────────────────────────────────────────────┐
│                 Hadoop の中核的な設計目標                     │
├──────────────────────────────────────────────────────────────┤
│ 1. スケーラビリティ(Scalability)                             │
│    水平スケール — ノードを追加して容量/スループットを増やす  │
│                                                              │
│ 2. 耐障害性(Fault Tolerance)                                 │
│    障害は正常 — レプリケーションと再試行でデータ/作業を守る  │
│                                                              │
│ 3. データローカリティ(Data Locality)                        │
│    演算をデータの近くへ — ネットワーク転送を最小化           │
│                                                              │
│ 4. 汎用ハードウェア(Commodity Hardware)                      │
│    安価なサーバー多数 — 高価な単一機材に依存しない           │
│                                                              │
│ 5. スループット優先(High Throughput, not Low Latency)        │
│    大容量バッチ処理に最適 — リアルタイム応答が目標ではない   │
└──────────────────────────────────────────────────────────────┘

┌───────────────────────────────────────────────────────────────┐
│   処理レイヤ (Processing)                                     │
│   MapReduce · Spark · Tez · Hive · Pig · Flink ...           │
├───────────────────────────────────────────────────────────────┤
│   リソース管理レイヤ (Resource Management)                   │
│   YARN — ResourceManager / NodeManager / ApplicationMaster   │
├───────────────────────────────────────────────────────────────┤
│   ストレージレイヤ (Storage)                                 │
│   HDFS — NameNode / DataNode / Block / Replication           │
└───────────────────────────────────────────────────────────────┘

ファイル: bigfile.log (380 MB)、ブロックサイズ 128 MB

   bigfile.log
   ┌──────────────┬──────────────┬──────────────┐
   │  Block A     │  Block B     │  Block C     │
   │  128 MB      │  128 MB      │  124 MB      │
   └──────────────┴──────────────┴──────────────┘

各ブロックはレプリケーション係数(replication factor、デフォルト3)
の分だけ複数の DataNode に複製されます。

   Block A ──▶ DataNode 1, DataNode 3, DataNode 5
   Block B ──▶ DataNode 2, DataNode 3, DataNode 6
   Block C ──▶ DataNode 1, DataNode 4, DataNode 6

┌─────────────────────────────────────────────────────────────┐
│                        NameNode (マスター)                   │
│  - ファイルシステムのネームスペース(ディレクトリ/ファイル木)を管理 │
│  - 各ファイルがどのブロックで構成されているかを記録          │
│  - 各ブロックがどの DataNode にあるかのマッピングを保持      │
│  - メタデータはメモリに常駐 (高速な照会のため)              │
└─────────────────────────────────────────────────────────────┘
        │            │            │            │
        ▼            ▼            ▼            ▼
   ┌─────────┐  ┌─────────┐  ┌─────────┐  ┌─────────┐
   │DataNode1│  │DataNode2│  │DataNode3│  │DataNode4│   (スレーブ)
   │ブロック保存│ │ブロック保存│ │ブロック保存│ │ブロック保存│
   └─────────┘  └─────────┘  └─────────┘  └─────────┘
   - 実際のブロックデータをローカルディスクに保存
   - 定期的に NameNode へハートビートとブロックレポートを送信

┌──────────── NameNode メタデータの永続化 ──────────────┐
│                                                       │
│   FsImage : 特定時点のファイルシステム全体のスナップショット │
│   EditLog : FsImage 以降に発生したすべての変更のログ │
│                                                       │
│   再起動時の復元手順:                                 │
│     1) FsImage をロード (基準スナップショット)        │
│     2) EditLog を再生(replay) (それ以降の変更を反映)  │
│     3) メモリ上の最新メタデータを完成                 │
│                                                       │
└─────────────────────────────────────────────────────────┘

        ┌──────────────┐   共有 EditLog     ┌──────────────┐
        │   Active     │◀────────────────▶│   Standby    │
        │   NameNode   │  (JournalNodes)   │   NameNode   │
        └──────┬───────┘                   └──────┬───────┘
               │                                  │
               │   DataNode は両方の NameNode に   │
               │   ハートビートとブロックレポートを送信 │
               ▼                                  ▼
        ┌───────────────────────────────────────────────┐
        │   DataNode 1 · DataNode 2 · DataNode 3 · ...   │
        └───────────────────────────────────────────────┘

   Active が死ぬ → ZKFC + ZooKeeper が検知 → Standby を Active へ昇格

[HDFS 書き込み経路]

  Client
    │ 1. create() 要求 — 「このパスにファイルを作りたい」
    ▼
  NameNode
    │   - 権限/存在確認、ネームスペースにファイル項目を作成
    │ 2. 最初のブロックを置く DataNode の一覧を返す (複製3 → 3個)
    │      例: DN1 ──▶ DN2 ──▶ DN3 (パイプライン)
    ▼
  Client
    │ 3. データをパケット単位で DN1 へ送信
    ▼
  ┌─────────┐  複製パイプライン   ┌─────────┐        ┌─────────┐
  │  DN1    │ ── パケット転送 ──▶│  DN2    │ ──────▶ │  DN3    │
  │ディスク↓│                    │ディスク↓│        │ディスク↓│
  └─────────┘                    └─────────┘        └─────────┘
    │                                                    │
    │ 4. ack は逆方向:  DN3 ──▶ DN2 ──▶ DN1 ──▶ Client     │
    ▼                                                    
  Client
    │ 5. ブロックが満杯になれば次のブロックのため 2〜4 を繰り返す
    │ 6. close() — NameNode に書き込み完了を通知
    ▼
  NameNode (メタデータを最終確定)

[デフォルトの複製配置 (replication = 3、ラック認識ベース)]

  複製1 : クライアントのいるノード(または同じラックの任意のノード)
  複製2 : 複製1 と異なるラックのノード
  複製3 : 複製2 と同じラックの別のノード

  目的:
   - 同じラックに2個 → ラック内ネットワークで高速書き込み
   - 別のラックに1個 → ラック全体の障害時にもデータを保全

[HDFS 読み取り経路]

  Client
    │ 1. open() — 「このファイルを読みたい」
    ▼
  NameNode
    │ 2. ファイルのブロック一覧 + 各ブロックの DataNode 位置を返す
    │    (クライアントに近い順に並べて渡す)
    ▼
  Client
    │ 3. 各ブロックごとに最も近い DataNode から直接読み取り
    ▼
  ┌─────────┐  ┌─────────┐  ┌─────────┐
  │  DN (A) │  │  DN (B) │  │  DN (C) │   ← ブロックごとに近いノードを選ぶ
  └─────────┘  └─────────┘  └─────────┘
    │
    │ 4. もしある DataNode が応答しなければ
    │    → 別の複製を持つ DataNode へ再試行
    ▼
  Client (ブロックを繋ぎ合わせて完全なファイルに)

┌──────────────────────────────────────────────────────────────┐
│                  ResourceManager (RM, マスター)              │
│   - Scheduler: クラスタのリソースをアプリに割り当てる        │
│   - ApplicationsManager: AM の起動/再起動を管理              │
└───────────────┬──────────────────────────────────────────────┘
                │ リソース割り当て / 状態報告
   ┌────────────┼────────────────────────┬───────────────┐
   ▼            ▼                        ▼               ▼
┌────────┐  ┌────────┐              ┌────────┐      ┌────────┐
│ Node-  │  │ Node-  │              │ Node-  │      │ Node-  │
│Manager │  │Manager │   ...        │Manager │      │Manager │
│ (NM)   │  │ (NM)   │              │ (NM)   │      │ (NM)   │
└───┬────┘  └───┬────┘              └───┬────┘      └───┬────┘
    │           │                       │               │
 コンテナ     コンテナ                コンテナ         コンテナ
 ┌──────┐   ┌──────┐               ┌──────┐        ┌──────┐
 │ App  │   │ Task │               │ Task │        │ Task │
 │Master│   │      │               │      │        │      │
 └──────┘   └──────┘               └──────┘        └──────┘

[YARN ジョブ投入および実行シーケンス]

  Client                RM              NM(複数ノード)       AM
    │                    │                   │               │
    │ 1. submitApplication                   │               │
    │───────────────────▶│                   │               │
    │                    │ 2. AM コンテナ割り当て要求         │
    │                    │──────────────────▶│               │
    │                    │                   │ 3. AM コンテナ │
    │                    │                   │    起動        │
    │                    │                   │──────────────▶│
    │                    │ 4. AM 登録(register)               │
    │                    │◀──────────────────────────────────│
    │                    │ 5. タスク用コンテナ要求            │
    │                    │◀──────────────────────────────────│
    │                    │ 6. コンテナ割り当て応答            │
    │                    │──────────────────────────────────▶│
    │                    │                   │ 7. AM が NM に │
    │                    │                   │   タスク実行を │
    │                    │                   │   指示         │
    │                    │                   │◀──────────────│
    │                    │                   │ 8. タスク実行  │
    │                    │                   │  (コンテナ内)  │
    │ 9. 進捗を照会      │                   │               │
    │◀───────────────────┼───────────────────┼───────────────│
    │                    │ 10. 完了後 AM 登録解除、リソース返却 │
    │                    │◀──────────────────────────────────│
    ▼                    ▼                   ▼               ▼

// Mapper: 1 行を受け取り、各単語について (word, 1) を放出
public class WordCountMapper
    extends Mapper<LongWritable, Text, Text, IntWritable> {

  private final static IntWritable ONE = new IntWritable(1);
  private final Text word = new Text();

  @Override
  protected void map(LongWritable key, Text value, Context context)
      throws IOException, InterruptedException {
    String line = value.toString();
    for (String token : line.split("\\s+")) {
      if (!token.isEmpty()) {
        word.set(token);
        context.write(word, ONE);
      }
    }
  }
}

// Reducer: 同じ単語に対する 1 をすべて足す
public class WordCountReducer
    extends Reducer<Text, IntWritable, Text, IntWritable> {

  private final IntWritable result = new IntWritable();

  @Override
  protected void reduce(Text key, Iterable<IntWritable> values, Context context)
      throws IOException, InterruptedException {
    int sum = 0;
    for (IntWritable v : values) {
      sum += v.get();
    }
    result.set(sum);
    context.write(key, result);
  }
}

// Driver: ジョブ設定と投入
public class WordCountDriver {
  public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    Job job = Job.getInstance(conf, "word count");
    job.setJarByClass(WordCountDriver.class);
    job.setMapperClass(WordCountMapper.class);
    job.setReducerClass(WordCountReducer.class);
    job.setCombinerClass(WordCountReducer.class); // マップ段階での部分集計
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(IntWritable.class);
    FileInputFormat.addInputPath(job, new Path(args[0]));
    FileOutputFormat.setOutputPath(job, new Path(args[1]));
    System.exit(job.waitForCompletion(true) ? 0 : 1);
  }
}

[MapReduce 全体のデータフロー (シャッフルを含む)]

  入力 (HDFS ブロック群)
   ┌────────┐ ┌────────┐ ┌────────┐
   │ Split1 │ │ Split2 │ │ Split3 │
   └───┬────┘ └───┬────┘ └───┬────┘
       ▼          ▼          ▼
   ┌────────┐ ┌────────┐ ┌────────┐
   │ Map 1  │ │ Map 2  │ │ Map 3  │     ← 各 Map が (キー,値) を放出
   └───┬────┘ └───┬────┘ └───┬────┘
       │          │          │
       │  [パーティション + ソート + (任意)コンバイナ]
       │          │          │
       └────┬─────┴─────┬────┘
            │  S H U F F L E  │           ← ネットワークでキー基準に再分配
       ┌────┴─────┐ ┌────┴─────┐
       ▼          ▼ ▼          ▼
   ┌────────────────┐ ┌────────────────┐
   │  Reduce 1      │ │  Reduce 2      │  ← 同じキーは同じ Reducer へ
   │ (キー A,B 担当) │ │ (キー C,D 担当) │
   └───────┬────────┘ └───────┬────────┘
           ▼                  ▼
       ┌────────┐         ┌────────┐
       │ 出力 1 │         │ 出力 2 │       ← HDFS に結果を書き込み
       └────────┘         └────────┘

[Map 側 → Reduce 側のシャッフル詳細]

  Map 側:
   1. map() 出力がメモリバッファに溜まる
   2. バッファが満杯になればディスクへスピル(spill) — このときパーティション別にソート
   3. 複数のスピルファイルを 1 つにマージ(merge)
   4. (コンバイナ設定時) マップ側で部分集計してデータ量を縮小

  Reduce 側:
   5. 各 Map から自分のパーティションのデータを取得(fetch、ネットワーク)
   6. 取得したデータをキー基準でマージ + ソート
   7. reduce() 呼び出し — 同じキーの値を一度に処理

┌──────────────────────────────────────────────────────────────┐
│  SQL / クエリ:  Hive, Impala, Presto/Trino                   │
│  スクリプト:    Pig                                          │
│  処理エンジン:  MapReduce, Spark, Tez, Flink                 │
│  NoSQL:         HBase (HDFS 上の分散キー値/カラムストア)      │
│  収集:          Sqoop(RDB-HDFS), Flume/Kafka(ストリーム)     │
│  調整:          ZooKeeper (分散コーディネーション)           │
│  ワークフロー:  Oozie, Airflow                               │
├──────────────────────────────────────────────────────────────┤
│  YARN  (リソース管理)                                        │
├──────────────────────────────────────────────────────────────┤
│  HDFS  (分散ストレージ)                                      │
└──────────────────────────────────────────────────────────────┘

-- 外部テーブル定義: HDFS パスのデータをテーブルのように扱う
CREATE EXTERNAL TABLE access_log (
  ip        STRING,
  ts        STRING,
  url       STRING,
  status    INT
)
ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'
STORED AS TEXTFILE
LOCATION '/data/access_log/';

-- ステータスコード別に集計
SELECT status, COUNT(*) AS cnt
FROM access_log
GROUP BY status
ORDER BY cnt DESC;

[HBase の位置づけ]

   アプリケーション (低レイテンシの単件照会が必要)
        │
        ▼
   ┌──────────┐
   │  HBase   │   ← ランダム読み書き、行単位アクセス
   └────┬─────┘
        ▼
   ┌──────────┐
   │  HDFS    │   ← 実際のデータファイル(HFile)を永続保存
   └──────────┘

[古典的な Hadoop]                 [クラウド/レイクハウス]

 ┌──────────────┐                 コンピュート(必要なときだけ)
 │ノード=ストレージ│               ┌──────┐ ┌──────┐ ┌──────┐
 │     + コンピュート│ (co-location)│Spark │ │Trino │ │Flink │
 │ HDFS  +  YARN  │                └───┬──┘ └───┬──┘ └───┬──┘
 └──────────────┘                     │        │        │
   データローカリティ中心             └────────┼────────┘
   ストレージ・コンピュート結合                ▼
                                       ┌────────────────┐
                                       │ オブジェクトストレージ │
                                       │  S3 / GCS / ... │  (分離)
                                       └────────────────┘
                                  ストレージ/コンピュートを独立スケール

[レイクハウスの階層構造]

 ┌──────────────────────────────────────────────┐
 │  クエリ/処理: Spark, Trino, Flink, Dremio ... │
 ├──────────────────────────────────────────────┤
 │  テーブルフォーマット: Iceberg / Delta / Hudi │
 │   - ACID トランザクション - スキーマ進化 - タイムトラベル │
 ├──────────────────────────────────────────────┤
 │  ファイルフォーマット: Parquet / ORC (カラム指向) │
 ├──────────────────────────────────────────────┤
 │  ストレージ: オブジェクトストレージ(S3/GCS) または HDFS │
 └──────────────────────────────────────────────┘

[よくある点検ポイント]

 - コンテナメモリ( yarn.scheduler.maximum-allocation-mb )が
   タスクに十分か? OutOfMemory でコンテナがキルされていないか
 - マップ/リデュースタスク数が適切か (少なすぎると並列性不足、
   多すぎるとスケジューリングのオーバーヘッド)
 - データスキュー(skew): 特定のキーに値が偏り、ある Reducer だけが長く動くか
 - シャッフル量が過剰でないか (コンバイナ/パーティショナで減らせるか)
 - キュー設定が適切か (特定のキューがリソースを独占していないか)