Sparkジョブを実行すると、実際には何が起きるのか？

はじめに

データセットが大きくなると、CPUの処理能力、メモリ、ストレージのスループットに制約があるため、単一のマシンで効率よく処理することが難しくなる場合があります。

Apache Sparkは、ワークロードをパーティションに分割し、CPUコアや複数のマシンにまたがってタスクを並列実行できる分散データ処理エンジンを提供することで、こうした課題に対処します。

ただし、Sparkは単にマシンを増やせばよいというものではありません。

ドライバー、エグゼキューター、RDD、DataFrame、パーティション、ジョブ、ステージ、タスクがどのように連携するかを理解することは、効率的で信頼性の高いデータパイプラインを設計するうえで欠かせません。

この記事では、eコマースの売上集計を例に、Sparkのアーキテクチャや実行の仕組み、パフォーマンスに関する重要な考慮事項を解説します。

1. Apache Sparkを使う理由 — 並列データ処理

Apache Sparkは、大規模なデータ処理とデータエンジニアリング向けに設計された分散コンピューティングエンジンです。

Sparkは、並列データ処理を活用してビッグデータを処理します。

図A. Apache Sparkの分散クラスターアーキテクチャの概要（クラスター、ドライバー、ワーカーノード）。単一ノードでの実行との比較（Kuriko IWAI作成）

単一ノードでの処理（右側、図A）と比べ、Sparkは作業をタスクに分割し、複数のノードを使って実行します。利用可能な場合は、マシンをまたいでリソースを活用してタスクを実行します。

ここでいう分散は、複数のマシンが必須という意味ではありません。Sparkは単一のマシン上でローカルモードでも実行できます。その場合でも、データをパーティションに分けて並列処理できます。

2. Apache Sparkの主要なアーキテクチャ

Sparkはいくつかの主要なコンポーネントで構成されています。

• Sparkドライバー（ドライバー）：Sparkアプリケーションの中心となって調整を行うプロセス。
• クラスター管理機能：CPUコアやメモリなどのハードウェアリソースを割り当てる外部サービス。
• ワーカーノード（ワーカー）：クラスター内の物理マシン、仮想マシン、またはコンピューティングノード。
• エグゼキューター：ワーカーノード上で起動される分散ワーカープロセス。
• Java Virtual Machine (JVM)：Sparkのバイトコードを実行するために必要なランタイム環境（SparkはScala/Javaで記述されているため）。

まず、ドライバーがSparkアプリケーションを調整し、実行タスクを表す有向非巡回グラフ（DAG）に変換して、エグゼキューターに割り当てます。

次に、クラスター管理機能が、デプロイ環境のリソース管理システムに基づいてリソースを割り当てます。

ワーカーノードはエグゼキューターにコンピューティングリソースを提供します。エグゼキューターはタスクを実行し、その実行中はデータを保持します。

Databricksなどの一部のマネージド環境では、1つのワーカーにつきエグゼキューターが1つ配置されますが、これはSpark全般に当てはまるルールではありません。クラスターの構成によっては、1つのワーカーで複数のエグゼキューターを実行できます。

2-1. eコマースのユースケース — 入力データの構造

小規模なeコマース企業を想定してみましょう。

この企業には10,000件の販売記録があり、それぞれに在庫管理単位（SKU）と販売数量が記録されています。SKUは3種類あるとします。 A, B、および C.

表1. eコマースの販売記録の内訳と集計予想の例。

目標は、SKUごとの有効な販売数量の合計を算出することです。

1. 次の条件に該当する記録を除外する qty <= 0、その後
2. SKUごとにサンプルを集計する。

説明用の全データセットでは、フィルタリング後に10,000件の記録のうち9,800件が残るとします。Sparkで得られると予想される最終合計は次のとおりです。

[("A", 8000), ("B", 10500), ("C", 9200)]

2-2. Sparkクラスター

このユースケースのために、簡略化したクラスターを設定します。

ワーカーノード（ワーカー）

ワーカーノードを3台用意し、それぞれを8 vCPU、RAM 32 GiBのインスタンスタイプにします。

図B. ワーカーノード間のリソース割り当て（Kuriko IWAI作成）

システムでは、合計24 vCPUを利用できます。

SparkのエグゼキューターとCPU／メモリの割り当て

ここでは、次のように確保するとします。

• エグゼキューターを7つ。
• 各エグゼキューターに2つのvCPUコア。
• 各エグゼキューターに3GiBのJVMヒープメモリ。

Sparkアプリケーションは、これら7つのエグゼキューターにリソースを割り当てるよう、クラスターマネージャーに要求します。

次に、クラスターマネージャーは各ワーカーのリソース使用状況に基づいて、CPUとメモリの割り当てを決定します。

たとえば、2 vCPUコア／エグゼキューター × 7エグゼキューターとして、タスクスロットを14個作成すると決定します。

図C-1. ワーカーAでリソースの偏りがある場合の、ワーカー間のvCPUコア、メモリ、エグゼキューターの割り当て

エグゼキューターに割り当てられる3 GiBのメモリが、エグゼキューターの総メモリ量ではない点に注意してください。ワーカーやほかの実行処理のために、メモリオーバーヘッドを確保する必要があります。

単一のタスクが1 vCPU（spark.task.cpus = 1）を必要とすると仮定すると、各ワーカーは自身のvCPUを次のように割り当てます。

図C-2. ワーカー間で並列に行われるvCPUの割り当て

ここから、次の2つの重要な点がわかります。

1. 各ワーカーのCPU使用率はそれぞれ異なります（図C-2では、ワーカーAの使用率は75％である一方、ワーカーBとCは50％のままです）。
2. システムには24 vCPUコア（1ワーカーあたり8 vCPU × ワーカー3台）がありますが、利用できるのは14 vCPUコアだけで、残りの10コアはこれらのタスクには使われません。

2-3. Sparkを支えるJava仮想マシン（JVM）

Apache Sparkは、Java仮想マシン（JVM）上に構築されています。

JVMは、Sparkのバイトコードを実行するために必要なランタイム環境です。

SparkはScala／Javaで書かれているため、次のようになります。

• アプリケーションでScalaまたはJavaを使用する場合：コードはJVM内でネイティブに実行されます。
• PySpark（Python）を使用する場合：Pythonプロセスがコードを実行しますが、Py4Jなどのブリッジを介して、JVMベースのドライバーと通信します。ワーカーノード上のSparkエグゼキューターは引き続きJVM内で動作し、そこでPythonワーカーを管理して、PythonコードやPythonユーザー定義関数（Python UDF）を実行します。

いずれの場合も、JVMはドライバーとエグゼキューターの実行スレッド、メモリ割り当て、ガベージコレクションを管理します。

PySparkアプリケーションにおけるドライバーとエグゼキューターのJVM

PySparkアプリケーションでは、PythonドライバーがJVM上で動作するSparkドライバーと通信します。

• Pythonドライバー：アプリケーションレベルのPySparkコードを記述し、実行する場所です。
• ドライバーJVM：Sparkの実行とスケジューリングを調整します。
• エグゼキューターJVM：Sparkタスクを実行し、実行リソースを管理します。タスクでPythonの実行が必要な場合は、Pythonワーカープロセスが関与します。
• クライアントJVM（クライアントプロセスまたはジョブ投入マシン）：ローカルマシンまたはエッジ／ゲートウェイノード上で動作するJVMです。

図に示すように、クライアントJVMはまずドライバーJVMに接続し、その後、ドライバーJVMがエグゼキューターJVMと連携します。

図D. クライアント、ドライバー、エグゼキューターの各JVMの全体的な関係

ここでは、ドライバーJVMがSparkドライバーをホストしてアプリケーションを管理し、エグゼキューターJVMが個々のエグゼキューターをホストして、それらのタスクを実行します。

各JVMドライバーには、データ処理、キャッシュ、ユーザーオブジェクトを効率よく扱うためのJVMヒープがあります。

2-4. Resilient Distributed Dataset（RDD）を詳しく解説

Resilient Distributed Dataset（RDD）は、Sparkが並列処理できる不変で、パーティションに分割された要素の集合です。

RDDは変換処理間の依存関係を追跡するため、多くの障害シナリオにおいて、失われたパーティションをSparkが再計算できます。

この名称には、次の3つの特性が表されています。

• 耐障害性：Sparkはリネージュ（系譜情報）、つまり、どのRDDがどの先行RDDに依存しているかを記録します。そのため、中間結果をすべて複製して保持する代わりに、失われたパーティションを再構築できる場合が多くあります。
• 分散：データセットはパーティションとして表現され、Sparkはそれらを異なるエグゼキューターで処理できます。複数の物理マシンは必須ではなく、ローカルモードでもRDDを利用できます。
• データセット：RDDには、キーと値のペア、整数、文字列、タプルなど、サポートされているさまざまな要素を含めることができ、リレーショナルな行だけに限られません。RDD APIには、次のような操作が含まれます map, filter, reduceByKey, join、および aggregateByKey.

RDDは論理的なデータ集合および計算の抽象化であり、特定のパーティションが特定のワーカーに恒久的に割り当てられることを宣言するものではありません。

パーティション

パーティションはデータを論理的に分割したもので、通常はステージ内の1つのタスクに対応します。

ここでは、100個のパーティションを作成するとします（この100個の初期パーティションは、説明のために明示的に選んだものです）。

入力がほぼ均等に分散されている場合、ソース要素が10,000個あれば、各パーティションには約100個の要素が含まれます。

図E. 100個の初期データ分割へのパーティションの並列割り当て

3. RDD APIとDataFrame API：Catalystによる内部最適化

最初のRDDの例は、リネージ、パーティション、シャッフルを理解するのに役立ちます。

一方、一般的な構造化ETLでは、列、スキーマ情報、SQLに似た式、オプティマイザーが認識できる操作を利用できるため、DataFrame APIから始めるのが望ましいです。

以下の図は、その関係を示しています。

図F. Sparkにおけるグラフの概念の関係：RDDリネージ、DataFrame、Spark実行DAG

RDDが変換を定義するためにRDDリネージを持つのと同様に、DataFrameのクエリプランは構造化クエリを表します。

Sparkは依存関係と物理演算子に基づいて実行処理を構成します。

DataFrameの実行では内部的にRDDベースの仕組みが使われることがありますが、だからといってDataFrameのコードを書く際に、ユーザーがRDDリネージを手作業で構築する必要があるわけではありません。

3-1. DataFrameのクエリプラン

DataFrameのクエリプランとは、実行エンジンが生成する論理的・物理的な処理ステップの連なりです（図Lの紫色のボックス）。

DataFrameのクエリは、次のようないくつかの表現を経て実行されます。

1. 論理プラン：要求された操作（スキャン、フィルタリング、グループ化、集計）の記述。
2. 解析・最適化済み論理プラン：Sparkが参照を解決し、列のプルーニングや述語のプッシュダウンなど、適用可能な論理最適化を行います。
3. 物理プラン：Sparkが実行可能な演算子（スキャン、部分集計・最終集計、交換、結合、ソートなど）を選択します。
4. 実行：物理プランはSparkの実行機構を用いて実行され、最終的にはエグゼキューター上でタスクが実行されます。

このユースケースでは、物理プランに次のステップが含まれる場合があります。

• 部分ハッシュ集計、
• SKUによる交換、そして
• 最終ハッシュ集計。

3-2. アクションの比較：RDD APIとDataFrame API

SKUの集計を例に、RDD APIとDataFrame APIを比較してみましょう。

RDD API

要素は (sku, qty) ペアです。Pythonでの集計ロジックは明示的に記述され、SparkはRDDの依存関係に従って処理します。

result_rdd = (
    raw_rdd
    .filter(lambda row: row[1] > 0)
    .reduceByKey(lambda a, b: a + b, numPartitions=4)
)

rdd_results = result_rdd.collect()

# Example:
# [('A', 8000), ('B', 10500), ('C', 9200)]

DataFrame API

次のように仮定します。 sales_df 列を持つ sku （文字列）および qty （数値）：

from pyspark.sql import functions as F

result_df = (
    sales_df
    .filter(F.col("qty") > 0)
    .groupBy("sku")
    .agg(F.sum("qty").alias("total_qty"))
)

df_results = result_df.collect()

# Example:
# [Row(sku='A', total_qty=8000),
#  Row(sku='B', total_qty=10500),
#  Row(sku='C', total_qty=9200)]

ここでは、どちらの例も同じビジネス上の課題を解決します。

無効な売上レコードを除外し、有効な数量をSKUごとに集計する。

また、どちらの場合も、 collect() Sparkの実行がトリガーされ、結果のすべてのレコードがDriverに返されます。

ここでの主な違いは、結果の表現方法です。

• RDD API：Pythonの要素を返します。たとえば、 (sku, total_qty) タプル。
• DataFrame API：構造化された Row 名前付きフィールドを持つオブジェクトを返します。たとえば、 sku および total_qty.

RDD APIとDataFrame APIの全体的な比較を以下に示します。

表2. RDD APIとDataFrame APIの機能・抽象化の比較：（Kuriko IWAI作成）

4. まとめ

Sparkのチューニングの目的は、パーティション数やCPU使用率を個別に最大化することではありません。

不安定さを招くことなく、エンドツーエンドの実行コストとレイテンシを削減することです。

この記事では、Sparkのシステムにおける中核的な関係を確認します。

• リソース：ワーカーはエグゼキューターをホストし、エグゼキューターはタスクを実行します。設定されたCPUとメモリの量が同時実行数を制限します。
• RDDプログラミング：変換によって不変のRDD定義とリネージが生成され、アクションによって計算が開始されます。
• DataFrameプログラミング：変換によって論理クエリ操作が生成され、最適化によって実行可能な物理演算子が選択されます。アクションによって計算が開始されます。
• 実行：Sparkは処理をジョブ、ステージ、タスクに編成し、タスクはエグゼキューター上でパーティションを処理します。シャッフルなどの依存関係によって、ステージ間に調整の境界が生じます。

Sparkが分散ワークロードをどのように実行するかを理解することは、あくまで基礎にすぎません。

RDD APIを使った7段階の計算フローを学ぶことができます。