pyspark REPLからの利用

Apache Sparkのダウンロードと展開

spark.apache.orgから、Apache Sparkをダウンロードします。

Downloads | spark.apache.org
https://spark.apache.org/downloads.html

このエントリの手順では「spark-2.4.3-bin-hadoop2.7.tgz」をダウンロードしました。

このファイルを適当な場所に展開します。

$ tar zxf spark-2.4.3-bin-hadoop2.7.tgz

以下のようにspark-shellを実行してversionが表示されることを確認します。

$ export SPARK_HOME=※展開先※/spark-2.4.3-bin-hadoop2.7
$ $SPARK_HOME/bin/spark-shell --version

pysparkの環境設定

次に、pysparkが利用するPythonを設定します。

numpyがインストールされたPython環境を利用する必要があるので、
ここではAnacondaのPython環境を指定します。

templateから、spark-env.shを準備します。

$ cp $SPARK_HOME/conf/spark-env.sh.template $SPARK_HOME/conf/spark-env.sh

spark-env.shの末尾に、以下のように利用するPythonのパスを記載します。

export PYSPARK_PYTHON=※ユーザのホームディレクトリ※/.anyenv/envs/pyenv/versions/anaconda3-2019.03/bin/python

# 上記はanyenvでAnacondaをインストールしている場合の例です。

起動とサンプルの実行

GraphFramesをパッケージに指定して、pysparkのREPLを起動します。

$SPARK_HOME/bin/pyspark --packages graphframes:graphframes:0.7.0-spark2.4-s_2.11

最新のパッケージは、spark-packages.orgで確認して下さい。

graphframes | SparkPackages
https://spark-packages.org/package/graphframes/graphframes

QuickStartにあるサンプルを実行してみます。

GraphFrames Quick-Start Guide
https://graphframes.github.io/graphframes/docs/_site/quick-start.html

以下のように結果が確認出来ます。

>>> v = sqlContext.createDataFrame([
...   ("a", "Alice", 34),
...   ("b", "Bob", 36),
...   ("c", "Charlie", 30),
... ], ["id", "name", "age"])
>>> e = sqlContext.createDataFrame([
...   ("a", "b", "friend"),
...   ("b", "c", "follow"),
...   ("c", "b", "follow"),
... ], ["src", "dst", "relationship"])
>>> from graphframes import *
>>> g = GraphFrame(v, e)
>>> g.inDegrees.show()
+---+--------+                                                                  
| id|inDegree|
+---+--------+
|  c|       1|
|  b|       2|
+---+--------+

IntelliJ IDEAでの開発環境設定手順

IntelliJ IDEAでGraphFramesの開発環境を設定する手順ですが、
以下エントリの手順でpyspark用の開発環境が設定出来ていることを前提として説明します。

IntelliJ IDEAでpyspark用の開発環境を設定する手順
https://takemikami.com/2018/11/27/IntelliJ-IDEApyspark.html

spark環境へのGraphFramesパッケージ追加

templateから、spark-defaults.confを準備します。

$ cp $SPARK_HOME/conf/spark-defaults.conf.template $SPARK_HOME/conf/spark-defaults.conf

spark-defaults.confの末尾に、
以下のように利用するGraphFramesパッケージを記載します。

spark.jars.packages    graphframes:graphframes:0.7.0-spark2.4-s_2.11

最新のパッケージは、spark-packages.orgで確認して下さい。

graphframes | SparkPackages
https://spark-packages.org/package/graphframes/graphframes

Python環境にgraphframesのインストール

Pythonの環境にGraphFramesをインストールします。

$ pip install graphframes

以上で、IntelliJ IDEAからGraphFramesを利用する環境が整いました。

サンプルコードの作成と実行

例えば、以下のようなコードを作成します。

pysparksample/graph.py

from pyspark.sql import SparkSession, SQLContext
from graphframes import GraphFrame

if __name__ == '__main__':
    spark = SparkSession.builder.appName('sample').getOrCreate()
    sqlContext = SQLContext(spark)

    v = sqlContext.createDataFrame([
        ("a", "Alice", 34),
        ("b", "Bob", 36),
        ("c", "Charlie", 30),
    ], ["id", "name", "age"])

    e = sqlContext.createDataFrame([
        ("a", "b", "friend"),
        ("b", "c", "follow"),
        ("c", "b", "follow"),
    ], ["src", "dst", "relationship"])

    g = GraphFrame(v, e)
    g.inDegrees.show()

「pysparksample/graph.py」コードの左側の三角記号から、処理を実行すると、以下のような結果が出力されます。

+---+--------+
| id|inDegree|
+---+--------+
|  c|       1|
|  b|       2|
+---+--------+