Apache Spark を Jupyter Notebook で試す (on ローカル Docker

以前 Spark を使ってたのですが今は使ってなくて，
そのうち忘れそうなので基本をメモしておくことにしました．

(全体的に聞きかじりの知識なので間違ってる点はコメント・編集リクエストを期待します)

## 使う

Jupyter + PySpark な環境が動く Docker イメージが用意されているので，ローカルで試すには便利です:
https://hub.docker.com/r/jupyter/pyspark-notebook/

PySpark とは，という話ですが，Spark 自体は Scala だけど，
Python で使えるやつがあってそれが PySpark だという話があります．

IPC でがんばってるという仕組みになっていたはずなので，
Scala <-> Python の変換のコストが結構でかいうんぬんみたいな話題もあります．

さて，使ってみましょう:

```sh
docker run -it -p 8888:8888 jupyter/pyspark-notebook
```

これを実行すると Terminal に 8888 番にトークンがついた URL が流れてくるので，
(`To access the notebook, ...` のあたり)
おもむろにアクセスすると Jupyter のページが出てきて，
Jupyter Notebook でコーディングできる簡単環境のできあがりです．

![Home_Page_-_Select_or_create_a_notebook.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/359439/78be505b-7c93-e922-9e69-62363e1ea9d0.png)

ここの `New` から `Notebook: Python3` を選択すれば Notebook を開けます

![Untitled_-_Jupyter_Notebook.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/359439/dc187674-39ab-c739-3330-3098fb73342f.png)

## 試す

動くかどうかのテストコードは以下で，サンプルからとってきました (https://jupyter-docker-stacks.readthedocs.io/en/latest/using/specifics.html#in-a-python-notebook):

```python
from pyspark.sql import SparkSession

spark: SparkSession = SparkSession.builder.appName("SimpleApp").getOrCreate()

# do something to prove it works
spark.sql('SELECT "Test" as c1').show()
```

`SparkSession` というやつはよくわからないけど，
Spark 自体のインスタンスみたいなものという認識です．

これを実行して表がでれば OK です:

![Untitled_-_Jupyter_Notebook.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/359439/6792a4f8-449c-c995-05a8-7413c3b79f50.png)

## データを扱う

こういうデータを対象にしてみましょう:

| `id` | `name` | `gender` | `age` |
|-----|---------|----------|--------|
| 1      | サトシ    | male        | 10        |
| 2     | シゲル     | male        | 10        |
| 3     | カスミ     | female    | 12        | 

### 入力と定義

Python で素朴にデータを定義するとこうなりますね:

```python
from typing import List, Tuple

Trainer = Tuple[int, str, str, int]
trainers: List[Trainer] = [
    (1, 'サトシ', 'male',   10),
    (2, 'シゲル', 'male',   10),
    (3, 'カスミ', 'female', 12),
]
```

各行の型は Python の `typing` でいう `Tuple[int, str, str, int]` となりますね．

で，Spark でもスキーマの定義があります:

```python
from pyspark.sql.types import StructField, StructType, StringType, IntegerType

trainers_schema = StructType([
    StructField('id',      IntegerType(), True),
    StructField('name',    StringType(),  True),
    StructField('gender',  StringType(),  True),
    StructField('age',     IntegerType(), True),
])
```

これで Spark 側での列のスキーマを定義できます．

Python で定義したデータを Spark の `DataFrame` に変換するにはこうします:

```python
from pyspark.sql import DataFrame

trainers_df: DataFrame = spark.createDataFrame(
    spark.sparkContext.parallelize(trainers),
    trainers_schema
)
```

これで `trainers_df` という `DataFrame` ができました．

データソースとして CSV とか MySQL とかそういうものから読み込めるので，
実際にはコード上で定義するよりそういうデータソースから読み込むことになるでしょう．
(場合により後述する JDBC とか，Hadoop の設定が必要です)

これをダンプして確認したい場合はこうします:

```python
trainers_df.show()
```

そうすると，表に整形されたテキストが数行出力されます:

![Untitled_-_Jupyter_Notebook.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/359439/315cc363-2804-fef0-d034-0f6131bfdbb8.png)

```
+---+------+------+---+
| id|  name|gender|age|
+---+------+------+---+
|  1|サトシ|  male| 10|
|  2|シゲル|  male| 10|
|  3|カスミ|female| 12|
+---+------+------+---+
```

### 集計と出力

ダンプではなく値をもらうには `.collect()` すればいいです:

```python
result = trainers_df.collect()
print(result)
```

CSV に書き出すときはこういう雰囲気で `DataFrame` を書き出します:

```python
trainers_df.coalesce(1).write.mode('overwrite').csv("path/to/output.csv")
```

入力同様，他にも S3, MySQL とか Elasticsearch とかいろいろ出力先がある雰囲気です．

`.coalesce(1)` はパーティションごとの分割されているデータを，
1つのパーティションに coalesce するというものです．
こうしないと，分割されたまま CSV 出力されます．

Hadoop の `hdfs` コマンドをつかって，
分割されたものを1つにまとめて取得するという手段もあります．

基本的に遅延評価になっていて，
`.collect()` みたいな操作をしてはじめて評価されるようになっているので，
そんなに頻繁に集計はしないはずです．

## 基本

これだけではただ表示しただけでまったく意味がないので適当な操作をしてみましょう:

```python
trainers_df.createOrReplaceTempView('trainers');

male_trainers_df = spark.sql('''
    SELECT *
    FROM   trainers
    WHERE  gender = 'male'
''')
male_trainers_df.show()
```

これはこういう結果を得ます:

| `id`|  `name`|`gender`|`age`|
|----|-----|------|---|
|  1|サトシ|  male| 10|
|  2|シゲル|  male| 10|

`DataFrame.createOrReplaceTempView(name)` は `DataFrame` を，
一時的な SQL の View として登録することができるものです．

これで `spark.sql(query)` で登録した View を対象に SQL の操作した結果の DF を得ることができるので，
こうすれば，全く臆することなく慣れ親しんだ SQL を使って Spark を使うことができて，
心理的障壁も学習コストも低いというマジックになっています．

View に登録しなくても，`DataFrame` のままコードで記述するという方法もあります:

```python
male_trainers_df = trainers_df.filter(trainers_df['gender'] == 'male')
```

こっちのほうが使いやすいケースもあるのでケースバイケースですね．

## 応用

SQL を使うことができるのだから，基本的な操作では別に問題ないのですが，
たいてい Spark を使いたいケースというのはなにかユーザー定義の操作をしたい状況になっていそうですね．

たとえば自分が過去やりたかったケースとしては，
「記事本文を形態素解析して分かち書きする」というものがあって，
これは SQL だけでは実現しがたいですね．

ただ，Python 上であれば MeCab があるので，
MeCab のライブラリを使って形態素解析してやれば何も考えなくても分解されてやってくるので，
僕のように全然わかってなくてもとりあえず MeCab に投げればいけるという手段をとれます．

そういう操作を Spark 上で `DataFrame` に対して行うにはどうすればいいかというと，
UDF (User-Defined Function) を定義するといいです．

(※ `DataFrame` ではなく RDD というものに対しては直接 `lambda` を適用できるという技がありますが，
 　 これはパフォーマンスが悪いというのがあります)．

UDF を定義するには次のような定義を行います:

```python
from pyspark.sql.functions import udf

@udf(StringType())
def name_with_suffix(name: str, gender: str) -> str:
    return name + {'male': 'くん', 'female': 'さん'}.get(gender, '氏')

spark.udf.register('name_with_suffix', name_with_suffix)
```

UDF となる関数に `@udf(ReturnType)` デコレーターを適用することで，
その関数は UDF として定義できるようになります．
それを Spark SQL で使うには `spark.udf.register(udf_name, udf)` して登録すれば，
`COUNT()` とかと同じ用にそのまま使えます．

ちなみにデコレーターを使わなくても，`udf_fn = udf(fn)` すれば既存の関数を適用できます．

この例としてあげたものは `gender` に応じて，
`name` に `gender` に応じた suffix をつけるというものです．
この関数を UDF として適用してみましょう:

```python
dearest_trainers = spark.sql('''
    SELECT name_with_suffix(name, gender)
    FROM   trainers
''')
dearest_trainers.show()
```

結果はこうなります:

|`name_with_suffix(name, gender)`|
|------------------------------|
|                    サトシくん|
|                    シゲルくん|
|                    カスミさん|

今回の例であれば SQL でも `CASE` を駆使して書けるというご意見がありますが，そのとおりです．

やりたいことによっては便利に使えるでしょう．

## UDF

さて，前述した形態素解析して分かち書きするというものですが，
これはイメージとしてこのような関数になります
(実際には MeCab をカッコよく使います):

```python
import re

# 半角/全角スペースや約物で文字列を分割する
@udf(ArrayType(StringType()))
def wakachi(text: str) -> List[str]:
    return [
        word
        for word
        in re.split('[ 　！…]+', text)
        if len(word) > 0
    ]
```

これを適用するのも同じくそのまま使えば OK です．
サンプルコードを今一度データを変更しつつ書いてみます:

```python
Trainer = Tuple[int, str, str, int, str]
trainers: List[Trainer] = [
    (1, 'サトシ', 'male',   10, 'ポケモン　ゲット　だぜ'),
    (2, 'シゲル', 'male',   10, 'このおれさまが せかいで いちばん！ つよいって ことなんだよ！'),
    (3, 'カスミ', 'female', 12, 'わたしの ポリシーはね… みず タイプ ポケモンで せめて せめて …せめまくる ことよ！'),
]

trainers_schema = StructType([
    StructField('id',      IntegerType(), True),
    StructField('name',    StringType(),  True),
    StructField('gender',  StringType(),  True),
    StructField('age',     IntegerType(), True),
])

trainers_df = spark.createDataFrame(
    spark.sparkContext.parallelize(trainers),
    trainers_schema
)
trainers_df.createOrReplaceTempView('trainers');

wakachi_trainers_df = spark.sql('''
    SELECT id, name, wakachi(comment)
    FROM   trainers
''')
wakachi_trainers_df.show()
```

ここでポイントになるのは，
今回の UDF は `str` を受け取って `List[str]` として展開するということですね．
これを実行してみるとこうなります:

|`id`|`name`|              `wakachi(comment)`|
|---|------|--------------------------------|
|  1|サトシ|        [ポケモン, ゲット, だぜ]|
|  2|シゲル|[このおれさまが, せかいで, い...|
|  3|カスミ|[わたしの, ポリシーはね, みず...|

展開されたセルはリストになっていて，
列のなかに更に列がある入れ子状態になっています．

これをそれぞれの `str` を列として展開したい場合どうすればいいかというと，
展開する関数を更に適用すればいいです:

https://spark.apache.org/docs/2.0.2/api/java/org/apache/spark/sql/functions.html#explode(org.apache.spark.sql.Column)

```python
from pyspark.sql.functions import explode

wakachi_trainers_df = spark.sql('''
    SELECT id, name, explode(wakachi(comment))
    FROM   trainers
''')
wakachi_trainers_df.show()
```

`explode` という関数があるので，
これを適用すれば入れ子になった要素がそれぞれの列として展開されます:

|`id`| `name`|          `col`|
|---|------|--------------|
|  1|サトシ|      ポケモン|
|  1|サトシ|        ゲット|
|  1|サトシ|          だぜ|
|  2|シゲル|このおれさまが|
|  2|シゲル|      せかいで|
|  2|シゲル|      いちばん|
|  2|シゲル|    つよいって|
|  2|シゲル|  ことなんだよ|
|  3|カスミ|      わたしの|
|  3|カスミ|  ポリシーはね|
|  3|カスミ|          みず|
|  3|カスミ|        タイプ|
|  3|カスミ|    ポケモンで|
|  3|カスミ|        せめて|
|  3|カスミ|        せめて|
|  3|カスミ|    せめまくる|
|  3|カスミ|        ことよ|

## ジョイン

さらなるポイントとして `DataFrame` どうしの `JOIN` ができます．
普通の MySQL とかの `JOIN` と変わらずに結合につかうカラムを指定して，
それをもとに `DataFrame` を結合するものです．

サンプルコードを更に追加して `JOIN` を使ってみます:

```python
Pkmn = Tuple[int, int, str, int]
pkmns: List[Pkmn] = [
    (1, 1, 'ピカチュウ', 99),
    (2, 1, 'リザードン', 99),
    (3, 2, 'イーブイ',   50),
    (4, 3, 'トサキント', 20),
    (5, 3, 'ヒトデマン', 30),
    (6, 3, 'スターミー', 40),
]
pkmns_schema = StructType([
    StructField('id',         IntegerType(), True),
    StructField('trainer_id', IntegerType(), True),
    StructField('name',       StringType(),  True),
    StructField('level',      IntegerType(), True),
])
pkmns_df = spark.createDataFrame(
    spark.sparkContext.parallelize(pkmns),
    pkmns_schema
)
pkmns_df.createOrReplaceTempView('pkmns');

trainer_and_pkmns_df = spark.sql('''
    SELECT     *
    FROM       trainers
    INNER JOIN pkmns
          ON   trainers.id = pkmns.trainer_id
''')
trainer_and_pkmns_df.show()
```

|`id`|`name`|`gender`|`age`|                           `comment`|`id`|`trainer_id`|   `name`|`level`|
|---|------|------|---|-----------------------------------|---|----------|----------|-----|
|  1|サトシ|  male| 10|             ポケモン　ゲット　だぜ|  1|         1|ピカチュウ|   99|
|  1|サトシ|  male| 10|             ポケモン　ゲット　だぜ|  2|         1|リザードン|   99|
|  3|カスミ|female| 12|  わたしの ポリシーはね… みず タ...|  4|         3|トサキント|   20|
|  3|カスミ|female| 12|  わたしの ポリシーはね… みず タ...|  5|         3|ヒトデマン|   30|
|  3|カスミ|female| 12|  わたしの ポリシーはね… みず タ...|  6|         3|スターミー|   40|
|  2|シゲル|  male| 10|このおれさまが せかいで いちばん...|  3|         2|  イーブイ|   50|

ちなみに `INNER JOIN`, `OUTER JOIN` の他に種類がいっぱいあります．
こちらの記事がわかりやすいので引用します:

https://qiita.com/ryoutoku/items/70c35cb016dcb13c8740

これで集合操作ができるので便利という感じです．

各 `JOIN` の概念はこのページのベン図がわかりやすいので引用します:

https://medium.com/@achilleus/https-medium-com-joins-in-apache-spark-part-1-dabbf3475690

ポイントとしてやはり `JOIN` はコストがかかっていて遅いです．
クラスタを組んでたとしたら，各所に分散したデータから見つけて `JOIN` して戻してとかそういう操作が行われているようです．

ので，後述するパフォーマンスチューニングが必要になってきます．

## パフォーマンス

現実のケースとして，膨大なデータセットと格闘するのはそうとう辛いものがあります．
というのも，4時間とかかかるものだったら，最後の方で落ちたらまたやり直しかとなって，
二回ミスると一日の業務時間を捧げたことになってしまって残業が確定します．

なので，そういうパフォーマンスを改善するために，`JOIN` の効率を上げるようにデータを削減したり，
パーティションの区切りかたを変えたり，
パーティションをなるべくクラスタ上に小さく断片化させないような工夫が必要です．

Broadcast Join というもので，あえて全クラスタにデータセットを重複して配置することで，
`JOIN` 時にデータセットの検索のコストを下げるとかそういうものもあります．

重要なテクニックとして，
各チェックポイント的なところで適宜 DataFrame を `.cache()` しておくことで，
パフォーマンスが劇的に改善されるというものもあります．

パフォーマンスについての公式ページを見るとそういうテクニックがあって参考になります:

https://spark.apache.org/docs/latest/sql-performance-tuning.html#broadcast-hint-for-sql-queries

## MySQL

さて，よくあるのが MySQL のデータベースから読み込んでうんぬんしたいというのがあります．
このケースでは MySQL を扱うために JDBC の MySQL コネクタを用意する必要がありますが，
こちらのかたのエントリと，その Docker イメージが参考になります:

- https://cloudfish.hatenablog.com/entry/2018/08/03/191424
- https://hub.docker.com/r/cloudfish/pyspark-notebook/dockerfile

しかしながら，MySQL は Spark では扱いづらいみたいなところもあります．
(いろいろハマりポイントがあります)

## 実際

Spark が威力を発揮するのは:

- データがとにかくでかい
- 適用したい処理が互いに依存しない
- 各操作に副作用がなくて内部の操作で完結する (外の API への操作とかがない)

というものだと思っています．

あと Spark は複数台でクラスタを作って worker に仕事をさせるのがキモなので，
現実的には AWS におまかせするということで，Amazon EMR とか AWS Glue でやるのがよさそうですね．
ローカルだとクラスタを作らずに動くので，本気の巨大データを打ち込んでもパフォーマンスは出ずに恩恵にはあずかれないからです．

メモリの限界にぶち当たったり，
節約できても処理全体にバッチ流して二週間かかりますみたいな巨大データだと，
素朴なものでも自前で分割して複数プロセスにわけて実行とかすればできるかもしれないけど，
Spark にできることなら任せるのもよさげです．