MatplotlibでPySpark SQLの結果をプロットする方法を徹底解説

Matplotlibを使ってPySpark SQLのクエリ結果をグラフ化する方法を解説します。Sparkで処理したデータをPandasデータフレームに変換し、Matplotlibで可視化まで行う一連の手順を、サンプルコード付きで紹介します。

実装の全体フロー

MatplotlibでPySpark SQLの結果をプロットするには、以下の手順に従います。

• 図のサイズを設定し、サブプロット間および周囲の余白（パディング）を調整します。
• Spark機能のメインエントリポイントとなるSparkContextのインスタンスを取得します。
• Hiveに格納されたデータと統合できるSpark SQLの派生クラスであるHiveContextのインスタンスを取得します。
• タプル形式でレコードのリストを作成します。
• ローカルのPythonコレクションを分散処理用のRDDに変換します。
• リスト内の各レコードをDBスキーマ（Rowオブジェクト）としてマッピングします。
• スキーマを持つDataFrameを作成し、「my_table」というテーブル名で登録します。
• テーブルにレコードを挿入（登録）します。
• SQLクエリを実行してレコードを取得します。
• 取得したレコードをPandasのデータフレームに変換します。
• name列をインデックスに設定してプロットします。
• show()メソッドで図を表示します。

サンプルコード

from pyspark.sql import Row
from pyspark.sql import HiveContext
import pyspark
import matplotlib.pyplot as plt

plt.rcParams["figure.figsize"] = [7.50, 3.50]
plt.rcParams["figure.autolayout"] = True

sc = pyspark.SparkContext()
sqlContext = HiveContext(sc)

test_list = [(1, 'John'), (2, 'James'), (3, 'Jack'), (4, 'Joe')]
rdd = sc.parallelize(test_list)
people = rdd.map(lambda x: Row(id=int(x[0]), name=x[1]))
schemaPeople = sqlContext.createDataFrame(people)
sqlContext.registerDataFrameAsTable(schemaPeople, "my_table")

df = sqlContext.sql("Select * from my_table")
df = df.toPandas()
df.set_index('name').plot()

plt.show()

実行結果

このコードを実行すると、「my_table」から取得したidとnameのデータがMatplotlibの折れ線グラフとして描画され、画面に表示されます。name列がX軸のラベルとして使われ、idの値がY軸にプロットされます。

補足：より新しいAPIについて

なお、HiveContextは古いAPIであり、現在のPySparkではSparkSessionの使用が推奨されています。SparkSessionを使えば、SparkContextやHiveContextを個別に生成する必要がなく、以下のように簡潔に書けます。

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("example").getOrCreate()
df = spark.createDataFrame(test_list, ["id", "name"])
df.createOrReplaceTempView("my_table")

result = spark.sql("Select * from my_table")
result.toPandas().set_index('name').plot()

import matplotlib.pyplot as plt
plt.show()

環境に応じて適切なAPIを選択してください。