単語カウントの実装例
Spark Core APIを使用した単語頻度カウントの実装方法を示します。
// SparkContextはSparkアプリケーションのエントリーポイント
val sourceData = sc.textFile("hdfs://data/sample.txt") // HDFSからテキストファイルを読み込み
val tokenized = sourceData.flatMap(sentence => sentence.split("\\s+")) // 各行を単語に分割
val paired = tokenized.map(word => (word.toLowerCase, 1)) // 単語を小文字に変換し、(単語, 1)のペアを作成
val aggregated = paired.reduceByKey((count1, count2) => count1 + count2) // 同じ単語の出現回数を集計
aggregated.collect().foreach(println) // 結果を収集して表示
簡潔な実装例：
sc.textFile("hdfs://data/sample.txt")
.flatMap(_.split("\\s+"))
.map(word => (word.toLowerCase, 1))
.reduceByKey(_ + _)
.collect()
核心コンセプトの解説
RDD（弾力的分散データセット）
Sparkの基本となるデータ構造で、不変の分散コレクションです。各RDDは複数のパーティションで構成され、各パーティションはクラスタのノードに分散配置されます。
DAG（有向非巡回グラフ）
RDD間の依存関係を表現するグラフ構造です。SparkはこのDAGを基に実行計画を最適化します。
ステージ（Stage）
DAGをシャッフル操作の有無に基づいて分割した実行単位です。シャッフルを必要としない変換は同一ステージにまとめられます。
タスク（Task）
ステージ内の各パーティションに対して実行される最小の処理単位です。パーティション数とタスク数は一致します。
ジョブ（Job）
アクション操作の実行によってトリガーされる一連の処理のまとまりです。
Sparkジョブ実行フロー
- spark-submitによりアプリケーションが提交され、SparkContextが初期化されます
- TaskSchedulerがMasterノードと通信し、アプリケーションを登録します
- Masterはリソーススケジューリングアルゴリズムに基づき、WorkerノードでExecutorを起動します
- 起動したExecutorがDriverに逆方向で登録を完了します
- アクション操作が実行されるたびに新しいジョブが生成されます
- DAGSchedulerがジョブを複数のステージに分割し、各ステージでTaskSetを作成します
- TaskSchedulerがTaskSet内のタスクをExecutorに分配して実行します
- Executorのスレッドプールがタスクを並列実行し、結果をDriverに返します
実行モードの比較
Sparkは複数の実行モードをサポートしています：
ローカルモード
単一マシンでの実行に適したモードです。開発・テスト環境で主に利用されます。すべてのプロセスが同一JVM内で実行されます。
スタンドアロンモード
Spark独自のクラスタマネージャを使用するモードです。Master-Workerアーキテクチャを採用し、Driverの配置場所に応じてClientモードとClusterモードが存在します。
# スタンドアロンクラスタへのジョブ提交例
./bin/spark-submit \
--class com.example.DataProcessor \
--master spark://cluster-master:7077 \
--deploy-mode client \
--executor-memory 4G \
--num-executors 10 \
/opt/spark-apps/data-processor.jar \
input-path output-path
YARN統合モード
Hadoop YARNをリソースマネージャとして利用するモードです。大規模なエンタープライズ環境で標準的に採用されています。
# YARNクライアントモード
./bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master yarn \
--deploy-mode client \
--executor-memory 2G \
--num-executors 5 \
--conf spark.yarn.queue=production \
examples.jar 1000
# YARNクラスタモード
./bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master yarn \
--deploy-mode cluster \
--executor-memory 2G \
--num-executors 5 \
--conf spark.yarn.queue=production \
examples.jar 1000
依存関係の種類と特性
ナロー依存（狭義依存）
親RDDの各パーティションが子RDDの最大1つのパーティションからのみ参照される関係です。パイプライン処理が可能で、同一ノード内での効率的な実行を支援します。
ワイド依存（広義依存）
子RDDのパーティションが親RDDの複数パーティションに依存する関係です。データの再配置（シャッフル）を伴うため、ネットワーク通信とディスクI/Oが発生します。
ナロー依存は障害回復時の再計算コストが低く、パフォーマンスに有利です。
Spark SQLのパフォーマンス優位性
Spark SQLがHiveより高速である理由：
1. 中間結果のメモリ保持
Hiveがシャッフル毎にHDFSへの書き込みを必須とするのに対し、Sparkは中間データをメモリにキャッシュ可能です。これによりディスクI/Oが大幅に削減されます。
2. 最適化された実行エンジン
CatalystオプティマイザとTungsten実行エンジンにより、クエリの論理最適化と物理実行計画の最適化が自動的に行われます。
3. JVM再利用によるオーバーヘッド削減
Hadoopが各タスクごとに新しいJVMプロセスを起動するのに対し、SparkはExecutor内でスレッドを再利用します。これによりJVM起動コストが削減されます。
4. インメモリ列フォーマット
Tungstenにより、CPUキャッシュにフレンドリーな列指向メモリフォーマットでデータを管理し、ベクトル化処理を可能にします。
アプリケーション配布のポイント
Sparkアプリケーションを配布する際の重要な設定：
Mavenビルド設定：
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<version>3.2.4</version>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<transformers>
<transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
<mainClass>com.example.SparkApplication</mainClass>
</transformer>
</transformers>
<filters>
<filter>
<artifact>*:*</artifact>
<excludes>
<exclude>META-INF/*.SF</exclude>
<exclude>META-INF/*.DSA</exclude>
<exclude>META-INF/*.RSA</exclude>
</excludes>
</filter>
</filters>
</configuration>
</execution>
</executions>
</plugin>