SparkをGoogle colaboratory上で試す

# 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が動作する仮想環境)が起動していないため、少し時間がかかりますが
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/150865/a5bc6b5d-6095-7b71-1102-9d6270ddfa34.png)

数秒で結果が表示されます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/150865/67163b51-868b-64b4-1535-43a2af57e622.png)  
標準のランタイムは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でのコードと同じです。

```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"以降が表示されれば準備完了です。

```python
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](https://earthquake.usgs.gov/earthquakes/feed/v1.0/csv.php)の地震データを利用します。

```
!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()は指定した行数だけ取り込んだデータを表示させることができます。

```python
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でのデータ操作  
代表的なデータ操作を試してみます。
    - カウント  
    データの件数を表示します。

    ```python
    rc.count()
    10658
    ```

    - SQLライクな操作  
    SQL構文に似たselect操作ができます。
    例では3つのカラムだけを指定し、表示させます。
    
    ```python
    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...)ので、まだまだ入り口ですがこの内容
であれば数分で試すことができました。