Apache SparkとPySparkの概要

大規模なデータ処理のためのオープンソース分散処理フレームワークのApache SparkとPythonでSparkを利用するためのインターフェースであるPySparkの概要について解説します。

Contents

1 Apache Spark
2 PySpark
- 2.1 PySparkの概要
- 2.2 PySpark環境
3 まとめ

Apache Spark

Apache Sparkの概要

Apache Spark（以降、Spark）は、大規模なデータを処理するためのオープンソースの分散処理フレームワークです。2009年にUC BerkeleyのAMPLabでMatei Zahariaによって開始されています。その後、2014年にDatabricks社が設立され、Sparkの開発と商用サポートを提供する企業として注目を浴びるようになっています。

分散処理フレームワークといえばApache Hadoopを思い浮かべる方がいるかと思います。Hadoopでは、大規模データに対するバッチ処理等では非常に効果的ですが、近年発展してきている機械学習等の頻繁にデータアクセスを繰り返すようなケースにはあまり適していませんでした。

そこで、Apache Sparkでは、メモリ上で処理を展開することで機械学習といった処理に対しても大規模データに対する分散処理ができるようにしました。

Apache Sparkの構成とAPI

Apache Sparkは、以下の4つのコンポーネントで構成されています。

Spark SQL：構造化データ処理
Spark Structured Streaming：ストリーム処理
MLlib：機械学習
GraphX：グラフ操作

各コンポーネントは、Spark CoreやSpark SQL Engineといった実行エンジンが支えており、Sparkは、「Scala」「SQL」「Python」「Java」「R」といったプログラミング言語のAPIを提供しています。

Apache Spark自体は、Scalaで記述されており、JVM(Java Virtual Machine)上で動作します。なお、Scalaは、Javaのオブジェクト指向に加えて関数型プログラミングに対応できるハイブリットなプログラミング言語のことです。

Spark SQL

Spark SQLは、構造データを処理するためのモジュールで、Sparkにおける中心的なコンポーネントです。Spark SQLにより、各種RDBMSやファイル(csv, text, json, parquet等)から簡単にデータを読み込んで扱うことができます。

Spark Structured Streaming

Spark Structured Streamingは、ストリーミング処理に対応して分散処理をするためのコンポーネントです。Apache Kafka (カフカ)といった分散ストリームプラットフォームのストリームデータと連携して処理をすることができるようになります。

MLlib

MLlibは、各種機械学習アルゴリズムを提供するコンポーネントです。回帰、分類、クラスタリングといった各種機械学習アルゴリズムを提供しており、機械学習のワークフローを分散処理で簡単に実行することができます。

GraphX

GraphXは、グラフ（ソーシャルネットワークグラフ等）を操作し、グラフの並列計算を実行するためのコンポーネントです。グラフを分析、接続、探索のための標準的なグラフアルゴリズムが提供されています。

Apache Sparkのエコシステム

Apache Hadoopでは、分散処理とストレージの両方を含むフレームワークでしたが、Apache Sparkはこれらを分割しています。この柔軟性により、HDFS、Apache Hive、Apache HBase、MongoDB、各種RDBMS等々といった多様なデータソースや、クラウドストレージのAzure Storage、Amazon S3等とも簡単に統合することができます。

PySparkのエコシスエム — Apache Sparkのエコシステム（公式ページから引用）
※内容は変わっている可能性があります。

上記は、Apache Sparkエコシステムに関連する各種フレームワークですが、機械学習、BI、ストレージ等における各種有名なフレームワークと連携できるようになっています。

Sparkアプリケーションの概念

Sparkは、例えばPythonであれば後述するPySparkといったAPIを使用することで簡単に使用することができ、Sparkが分散処理を自動的に実行してくれます。これによりプログラマの負担は非常に少なくなります。

ただし、Sparkの概念を理解せずに使用すると、非効率なコードになったり、チューニングが難しくなってしまいます。

Sparkのアプリケーションの概念は少し難しく、まとめると長くなってしまうため「Sparkアプリケーションの概念を理解する」というページにまとめています。興味があれば参考にしてください。

PySpark

PySparkの概要

Sparkの説明で、Sparkは各種プログラミング言語のAPIを提供していることを紹介しました。Pythonで、Sparkを扱うためのAPIをPySparkといいます。PySparkを使用することで、上記で紹介してきたようなSparkの各種コンポーネントの機能を簡単に使用することができるようになります。

本記事では、具体的な使い方などの詳細は説明しませんが、イメージを持ってもらうために簡単なサンプルコードを示します。

from pyspark.sql import SparkSession
from pyspark.sql.functions import lit

# Sparkセッションの作成
spark = SparkSession.builder.appName("ParquetExample").getOrCreate()

# サンプルのDataFrameを作成
data = [("Alice", 34), ("Bob", 45), ("Charlie", 29)]
columns = ["Name", "Age"]

df = spark.createDataFrame(data, columns)

# 新しい列を追加
df_with_country = df.withColumn("Country", lit("USA"))

# データフレームをParquetファイルに保存
df_with_country.write.parquet("output.parquet")

# Parquetファイルからデータを読み込む
parquet_data = spark.read.parquet("output.parquet")

# 保存したデータを表示
parquet_data.show()

【実行結果】
+-------+---+-------+
|   Name|Age|Country|
+-------+---+-------+
|Charlie| 29|    USA|
|  Alice| 34|    USA|
|    Bob| 45|    USA|
+-------+---+-------+

上記例では、SparkSessionで、Sparkのセッションを開始し、サンプルのDataFrameを用意、新しい列を追加してから保存しています。その後、保存したデータを読み込み、showで表示するような例になっています。

上記処理では、用意しているSpark環境によって分散されて処理が実行されます。

Note

Pythonに慣れている人は、DataFrameと聞くとpandasを思い浮かべるかと思います。SparkのDataFrameは、pandasのDataFrameと類似はしていますが、似て非なるものということは理解しておいてください。

PySparkで処理を実装する際にpandasのDataFrameに変換する方法もありますが、pandasのDataFrameでは、単一ノードでの処理になってしまい適切に分散がされなくなります。pandasのDataFrameに似た操作方法でプログラミングして分散処理させたい場合には、spark.pandasの使用を検討できます。

PySpark環境

PySparkの学習をしたい際に、お手軽に環境を用意する場合には、Docker環境を使って構築するのが簡単です。Docker環境を用いて、PySparkを使えるようにする方法については、「PySparkの実行環境をDockerで用意する方法」でまとめていますので興味があれば参考にしてください。

実際にPySparkを使う場合は、Sparkクラスタを物理サーバーなどを用いて構築することになります。また、主要なクラウド環境では、Spark環境を使えるようなPaaSを提供してくれていますので、それらを利用することで環境構築の負担を軽減することが可能です。（例えば、AWSのAmazon EMR、Microsoft AzureのSynapse Sparkプール等）