はじめに

今回から2回に渡って，並列データ処理系のひとつであるSparkについて解説します。まずはじめに，Sparkの開発が始められた経緯を紹介し，次にSparkの特徴を説明します。

Sparkが登場した背景

Sparkは，Hadoop MapReduceと同様に，複数の計算機を用いてデータ処理を行う並列データ処理系です。2009年に，カリフォルニア大学バークレー校のAMPLabにて，Matei Zaharia氏を中心として開発が始まりました。Sparkの開発が始まった当時，世の中にはすでにHadoopが存在しており，高い耐障害性を有しかつスケーラブルな並列データ処理を，コモディティな計算機を用いて行うことは一般的になりつつありました。しかし，Hadoop MapReduceは必ずしも個々の計算機のメモリを効率的に活用する設計ではありませんでした。

Hadoop MapReduceは，ジョブの実行においては，処理対象のデータを読み出す際，もしくは処理結果を出力する際に，HDFSなどの外部ストレージにアクセスします。このような方法は，たとえば次のような場合に，ジョブを高速に実行できない可能性があります。

1．複雑なデータ処理を行うために，複数のジョブを連ねて実行する場合

Hadoop MapReduceのジョブは，MapとReduceの2つの処理オペレータをこの順に接続して実行するため，複雑なデータ処理を行うためには複数のジョブを組み合わせる必要があります。その際，Mapはストレージからの読み出しを行うことを前提とし，Reduceはストレージへの書き出しを行うことを前提としているため，ジョブ間での中間データの受け渡しにはストレージの入出力を行う必要があります。この方法は，ストレージとしてHDFSを用いる場合においては，中間データはHDFSの機能によって複数の計算機上にレプリカが作られるため，耐障害性の面で利点があります。しかし，連なるジョブの数が多くなるにつれて，HDFSへの中間データの入出力に伴うI/O処理やシリアライズ・デシリアライズ処理，そしてレプリケーション処理にかかる実行時間が処理全体の実行時間に対して大きな割合を占める可能性があります。

2．同じデータを複数のジョブから利用する場合

Hadoop MapReduceは，処理対象のデータを複数回読み出す場合においても，当該データを繰り返しストレージから読み出します。すなわちHadoop MapReduceは，アクセスの局所性が高いデータの読み出しを効率化するための仕組みを有していません。たとえば，分析処理においては，分析業務における試行錯誤の過程で，同じデータに対してさまざまな切り口でアドホックに分析処理を実行することが多いですが，Hadoop MapReduceを用いる場合は，分析処理ごとにジョブを実行する必要があるため，同じデータを処理する場合であっても，それぞれのジョブが独立にデータを読み出します。

このほか，機械学習アルゴリズムを実行するジョブでは，結果が収束する，もしくは，一定の閾値以内に収まるまで同じデータを読み出して繰り返し処理を行うため，上記1.と2.の両方が当てはまります。

このようなデータ処理の高速性を低下させる問題を回避するための解決策としては，中間データやアクセスの局所性が高いデータを，計算機のメモリ上にキャッシュしておく方法が考えられます。そのようなアイディアを実装したフレームワークとして，繰り返し処理に特化したHaLoop（［1］）や，グラフ処理フレームワークPregel（［2］）などが提案されましたが，いずれも特定の用途に特化したデータ処理系でした。これに対してSparkは，複数の計算機のメモリを粗粒度の共有メモリとして抽象化し，当該共有メモリを操作する数十種類のインターフェースを用いることにより，上記問題のみならず多様なデータ処理を高効率に処理することを目指した並列データ処理系として開発が始められました（［3］）。

Sparkの抽象データ構造の利点

Sparkでは，処理対象のデータをRDD（Resilient Distributed Datasets）と呼ばれる分散コレクション（分散共有メモリ）として抽象化し，すなわち，メモリ上のレコードをRDDの要素として扱い，複数の計算機上でRDDを処理することで並列データ処理を実現します。RDDにおいては，多くの言語に具備されているコレクションに対する操作に類似した操作（オペレータ）が定義されています。つまり，Sparkは分散コレクションを手続き的に操作するインターフェースを有する並列データ処理系であるといえます。

Sparkには，次の利点があります。

1．並列データ処理をシンプルに記述できる

アプリケーション開発者は，各種プログラミング言語でコレクションを扱うかのように並列データ処理を記述できます。

2．汎用的な並列データ処理系として利用できる

事前に定義されているRDDに対する数十種類のオペレータにより，多様なデータ処理を記述できます。

RDDは汎用的な分散コレクションとして見ることができるため，Sparkのコアをラップする特定の用途に特化した並列データ処理ライブラリを開発することは比較的容易です。この特徴を活用し，Sparkには標準でグラフ処理や機械学習，ストリーミング処理などの用途向けのライブラリが付属しています。

また，RDDは実際にはパーティション関数で複数の計算機に分割されています。そのため，処理対象のデータサイズや件数が増加する場合でも，複数の計算機上でパーティション並列性（第5回）を活用したスケーラブルな処理が可能です。

複雑なデータ処理を効率的に行うためのデータ処理モデル

Sparkでは，map()やfilter()などのコレクションに対するオペレータのように，RDDを入力としてRDDを出力するオペレータがあらかじめ定義されています。アプリケーション開発者は，これらのオペレータを組み合わせて，RDDに対する一連の処理を記述し，ジョブを構成します。したがって，Sparkで複雑なデータ処理を行う場合には，必ずしも複数のジョブを組み合わせる必要がありません。そのため，Hadoop MapReduceであったような，ジョブ間で中間データを受け渡すためのストレージアクセスなどを削減することができ，Hadoop MapReduceと比較してデータ処理を高速に実行できる可能性があります。

計算機上でのRDDの処理

Sparkでは，計算機上で動作するエグゼキュータと呼ばれるプロセスが，部分ジョブ（タスク）を実行することで，アプリケーション開発者によって定義されたRDDの処理を並列に実行します。タスクは，RDDに対するmap()やfilter()などの一連の処理を，次に示す2種類の境界（図1）で分割した処理単位です。

1.RDDのパーティション（図1「タスクの境界（1）」）

タスクに含まれるオペレータのインスタンスは，RDDのパーテョションごとにコピーされるため（※1），パーティション並列性を活用できます。

2.集約処理を始めとするパーティションをまたいだデータシャッフル処理（図1「タスクの境界（2）」）

1つのタスクには，map()やfilter()などの処理が複数含まれます。そのため，同一タスク内では，各処理オペレータのインスタンス間の中間データの受け渡しは，外部ストレージへのアクセスや計算機間のネットワークI/Oを伴うことなく，タスクが実行された計算機のメモリを介して行うことができます。

※1）

エグゼキュータには，同じRDDの異なるパーティションに対するタスクが2回以上割り当てられる場合があります。その場合でもタスクを効率的に実行できるよう，実際の実装においてはタスクにはオペレータの命令列そのものは含まれておらず，命令列への参照と，オペレータを適用すべきパーティションのインデックスが含まれています。命令列は，ジョブ実行のタイミングで，一度だけエグゼキュータにコピーされます。

おわりに

今回は，Sparkが登場した背景と，Sparkにおける抽象データ構造およびデータ処理の特徴を説明しました。次回は，Sparkのさらなる高効率化の仕組みと耐障害性を実現する仕組みを説明します。

関連文献

［1］Yingyi Bu, Bill Howe, Magdalena Balazinska, Michael D. Ernst, “HaLoop: Efficient Iterative Data Processing on Large Clusters”

［2］Grzegorz Malewicz, Matthew H. Austern, Aart J. C. Bik, James C. Dehnert, Ilan Horn, Naty Leiser, and Grzegorz Czajkowski, “Pregel: A System for Large-Scale Graph Processing”

［3］Matei Zaharia, Mosharaf Chowdhury, Tathagata Das, Ankur Dave, Justin Ma,Murphy McCauley, Michael J. Franklin, Scott Shenker, Ion Stoica, “Resilient Distributed Datasets: A Fault-Tolerant Abstraction for In-Memory Cluster Computing“