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 Data Lake Storage、Amazon S3等とも簡単に統合することができます。

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

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

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

Sparkでは、後述するPython用のPySpark等、各言語用のAPIを使用することで簡単に使用することができ、分散処理を自動的に実行できます。これによりプログラマが分散処理を実装する負担は非常に少なくなります。ただし、Sparkの概念を全く理解せずに使用すると、非効率なコードになったり、チューニングが難しくなったりしてしまいます。

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

PySpark

PySparkの概要

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

本記事では、PySparkで提供されている各種機能について具体的な説明はしませんが、イメージを持ってもらうために簡単なサンプルコードを紹介します。

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環境によって分散されて実行されます。

本記事ではPySparkの使い方について細かな紹介はしませんが、以下のような各種実装に関する説明記事も作成していますので興味があれば参考にしてください。

Note

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

PySparkで処理を実装する際にpandasのDataFrameに変換する方法もありますが、pandasのDataFrameは、単一ノードで処理されるため効率的な分散処理ができませんので注意が必要です。

pandasのDataFrameに似た操作方法でプログラミングして分散処理させたい人もいることから、spark.pandasというものも用意されています。もし、pandasで既に実装されているプログラムをSpark環境へ移行する場合には使用を検討してみてください。

PySpark環境

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

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