Sparkを使ってみる

Sparkを触ってみたので記録を残しておきます。
環境はAzure上のUbuntuやローカルのCentOS7など試したのですが、selectを実行すると
エラーになるためたどり着いた環境がGoogle colaboratoryでした。
普通のJupyter notebookなどでも動作できると思います。

1. Google Colaboratory

Google colaboratoryとは何かというとGoogleがホストしているJupyter Notebookです。
Pythonコードを試して見たい時に環境構築することなくブラウザ上でコードを書いて
実行ボタンだけで動作する無料のツールです。
GPUも使用できるため機械学習を簡単に試すことができます。
ただし、ランタイムが12時間でshutdownされランタイム上のファイルが消えます。

1. ログイン  
[Google colaboratory](https://colab.research.google.com/notebooks/welcome.ipynb?hl=ja)にアクセスし、Googleアカウントでログインする。  
1. 新しいノートブックの作製  
メニューの"ファイル"から"Python3の新しいノートプック"を選ぶと新規ノートブックが作製されるので

コードを書いて左側のボタンを押して実行します。

始めはランタイム(=Pythonが動作する仮想環境)が起動していないため、少し時間がかかりますが

数秒で結果が表示されます。

標準のランタイムはPython3系のGPU/TPU無しですが、メニューの"ランタイム"→"ランタイムのタイプの変更"
からPython2系、GPU、TPUが選択できます。

2. Spark

Apache Sparkとは
一言で言うのが難しいパッケージですがよく言われるのが

分散処理フレームワーク

です。巨大なデータをたくさんの分析する人(Workerノード)が分担して高速に処理します。
Hadoopも同じようなフレームワークですが違いとしてはHadoopの後に作られただけあって、
Hadoopが苦手な高速処理を得意とします。
ただし、メモリに乗らない超巨大データのようなSparkが苦手とする分野もあるため棲み分けが必要です。
Sparkの機能というよりはSparkでCSVを簡単に扱えることを試してみたいと思います。

環境準備
作製したノートブックに環境を準備します。
Sparkモジュールインストール
通常のOSにインストールする際と同じ様にapt-getよりOpenjdkをインストール、Sparkは本家より
必要なファイルをダウンロードします。Pythonコードではなくシステムコマンドですので"!"を付けて実行します。

!apt-get update
!apt-get install openjdk-8-jdk-headless -qq > /dev/null
!wget -q http://archive.apache.org/dist/spark/spark-2.3.1/spark-2.3.1-bin-hadoop2.7.tgz
!tar xf spark-2.3.1-bin-hadoop2.7.tgz
!pip install -q findspark

Pythonの環境設定
JAVAとSparkの場所を指定します。このあたりはPythonでのコードと同じです。


os.environ["JAVA_HOME"] = "/usr/lib/jvm/java-8-openjdk-amd64"
os.environ["SPARK_HOME"] = "/content/spark-2.3.1-bin-hadoop2.7"

Findspark、pysparkモジュールインポートとセッション作製
Sparkを使う際はSpark Contextを経由してリソースを利用します。そのため、Spark Contextと
Spark Sessionを指定します。
sparkコマンド実行後に "SparkSession - in-memory"以降が表示されれば準備完了です。

import findspark
findspark.init()
from pyspark import SparkContext
sc = SparkContext.getOrCreate()

import pyspark
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate() 
spark

SparkSession - in-memory
SparkContext
Spark UI
Version
    v2.3.1
Master
    local[*]
AppName
    pyspark-shell

データ取得
Sparkで扱うデータを取得します。何でも良いので適当なOpen Data系のCSVを持ってきます。
今回はUSGSの地震データを利用します。

!wget https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/all_month.csv
--2019-12-18 02:28:20--  https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/all_month.csv
Resolving earthquake.usgs.gov (earthquake.usgs.gov)... 99.84.194.2, 99.84.194.83, 99.84.194.44, ...
Connecting to earthquake.usgs.gov (earthquake.usgs.gov)|99.84.194.2|:443... connected.
HTTP request sent, awaiting response... 200 OK
Length: unspecified [text/csv]
Saving to: ‘all_month.csv’

all_month.csv           [             <=>    ]   1.87M   689KB/s    in 2.8s    

2019-12-18 02:28:24 (689 KB/s) - ‘all_month.csv’ saved [1965927]

データ取り込み
ダウンロードしたCSVファイルをSparkで処理できるように取り込みます。
.show()は指定した行数だけ取り込んだデータを表示させることができます。

rc = spark.read.csv('all_month.csv',header=True)
rc.show(5)

+--------------------+----------+------------+-----+----+-------+----+---+-------+----+---+----------+--------------------+--------------------+----------+---------------+----------+--------+------+---------+--------------+---------+
|                time|  latitude|   longitude|depth| mag|magType| nst|gap|   dmin| rms|net|        id|             updated|               place|      type|horizontalError|depthError|magError|magNst|   status|locationSource|magSource|
+--------------------+----------+------------+-----+----+-------+----+---+-------+----+---+----------+--------------------+--------------------+----------+---------------+----------+--------+------+---------+--------------+---------+
|2019-12-18T02:20:...|   35.6925|-117.4803333| 4.51|0.97|     ml|  16|131|0.09322|0.21| ci|ci39234096|2019-12-18T02:24:...|11km SW of Searle...|earthquake|            0.5|       1.1|   0.198|     9|automatic|            ci|       ci|
|2019-12-18T02:15:...|33.4356667|    -116.949|10.38|1.35|     ml|  44| 40|0.08343|0.21| ci|ci39234088|2019-12-18T02:26:...|8km W of Aguanga, CA|earthquake|           0.22|      0.62|     0.1|    27|automatic|            ci|       ci|
|2019-12-18T02:03:...|38.8400002|-122.8793335| 2.26|1.05|     md|  16|138| 0.0215|0.04| nc|nc73316041|2019-12-18T02:11:...|12km ENE of Clove...|earthquake|           0.36|      0.66|    0.13|     4|automatic|            nc|       nc|
|2019-12-18T01:56:...|  -29.7521|   -178.5187|   10| 5.9|    mww|null| 62|   0.72|0.56| us|us60006szl|2019-12-18T02:19:...|80km SW of Raoul ...|earthquake|            6.3|       1.9|   0.071|    19| reviewed|            us|       us|
|2019-12-18T01:51:...|38.5631676|-122.2906647| 2.24|1.77|     md|   8| 99|0.03139|0.04| nc|nc73316036|2019-12-18T02:00:...|14km E of Angwin, CA|earthquake|           0.38|      0.86|    0.11|     7|automatic|            nc|       nc|
+--------------------+----------+------------+-----+----+-------+----+---+-------+----+---+----------+--------------------+--------------------+----------+---------------+----------+--------+------+---------+--------------+---------+
only showing top 5 rows

時刻、緯度、経度、マグニチュードなどのデータが取り込まれていることがわかります。

Sparkでのデータ操作
代表的なデータ操作を試してみます。

カウント
データの件数を表示します。

rc.count()
10658

SQLライクな操作
SQL構文に似たselect操作ができます。
例では3つのカラムだけを指定し、表示させます。

rc.select('mag', 'net', 'depth').show(5)
+----+---+-----+
| mag|net|depth|
+----+---+-----+
|0.97| ci| 4.51|
|1.35| ci|10.38|
|1.05| nc| 2.26|
| 5.9| us|   10|
|1.77| nc| 2.24|
+----+---+-----+
only showing top 5 rows

3.まとめ

使ってみての感想ですがGoogle Colaboratoryは簡単にPython環境を試すことができるツールでした。
操作せずに放置するとランタイムが切断されてしまいますが、コードはGoogle Driveに保存されるため
すぐに環境を戻すことができます。

また、Sparkもなんとなく聞いたことがある程度の認識でしたが、触ってみることで何を目的としたパッケージ
なのか雰囲気を掴むことができました。
もっとSparkらしいMapやFilterまで書けなかった(TypeError...)ので、まだまだ入り口ですがこの内容
であれば数分で試すことができました。