データ分析のための並列処理ライブラリDask

この記事は、Brainpad Advent Calender 15日目の記事です。  
本記事では、メモリに乗らないようなデータもPandasやNumPyライクに操作を行い、スケールアップ・スケールアウトにも対応できるライブラリ、Daskについて、簡単に紹介をします。

# はじめに
Pythonでデータ分析や機械学習をする際、PandasやNumPyを用いる場面が非常に多くなってきました。
しかし、PandasやNumPyではメモリに乗らないデータの扱いが難しかったり、基本的にシングルコアでの処理を行うため、速度が遅い、といった問題があります。例えば、サーバー上で実行する際、CPUの論理コアが32個あっても、1個のCPUしか使用していない、といった感じです。

[^1]: [pandsa0.21のドキュメント](https://pandas.pydata.org/pandas-docs/stable/enhancingperf.html)を見ると、eval関数の内部で、numexprが利用されており、並列処理を行う事はあるようですが、基本的にPandasの標準関数で並列処理をすることは出来ないようです。

近年、データ分析関連のライブラリは非常に多様化しており、派閥（?）が沢山あるようです。
個人的には、Pandas作者であるWes McKinney氏が提唱している[Pandas2.0](https://pandas-dev.github.io/pandas2/)に期待をしていますが、鋭意製作中との事で、まだ実際に使用する事が出来ないです。

本記事では、実際にすぐに使えて、学習コストが低い、という点で、Daskというライブラリで、並列処理を用いた分析を行ってみようと思います。
ちなみに、本記事では紹介をしませんでしたが、Daskは[分散処理](https://github.com/dask/dask-tutorial/blob/master/04_distributed.ipynb)を行ったり、[既存の機械学習ライブラリに、高速化](https://jcrist.github.io/dask-sklearn-part-1.html)を適用することが出来ます。


**本記事で行う事**

- Daskの簡単な紹介
- Dask.Arrayでの並列処理時のメモリ、実行時間比較
- Dask.DataFrameでの並列処理時のメモリ、実行時間比較

# Daskとは
## 概要
Daskは、柔軟に並列処理・分散処理を行う分析ライブラリです。
NumPyやPandasと競合するライブラリではなく、それらを高機能にしたラッパーライブラリのようになっています。
Daskの使用用途は大きく二つあります。

1. NumPy/Pandas/Listのような操作感で、メモリに乗らないような大きいデータ（Out-of-Core）の解析（DB、Sparkなどに近い用途）
2. カスタムタスクスケジューリング（Luigi, Airflow, Celery, or Makefilesなどに近い用途）

また、Daskのスケーリング方法も大きく二つに分かれます。

1. スレッド、または、プロセスを使用する単一マシンでの並列処理（スケールアップ）
2. 複数ノードでのクラスターによる並列処理（スケールアウト）

あまり知られていないようですが、分散処理も可能です。
本記事では、**シングルマシンでの並列処理について**、フォーカスしています。

Daskで主に使われる以下の3つのクラスは、別オブジェクトの要素から成る集合で構成されています。

| 集合          | 要素             |
|:--------------|:-----------------|
| Dask.Array    | NumPy            |
| Dask.DatFrame | Pandas.DataFrame |
| Dask.Bag      | PythonのObject   |

例えば、Dask.DataFrameは、Pandas.DataFrameの要素で構成されています。
<div align="center">
<img src="http://dask.pydata.org/en/latest/_images/dask-dataframe.svg" width="300">
</div>

> [Dask Document](http://dask.pydata.org/en/latest/)より

## 立ち位置
|              | メリット                                                                                                           | デメリット                                                                               |
|:-------------|:-------------------------------------------------------------------------------------------------------------------|:-----------------------------------------------------------------------------------------|
| NumPy/Pandas | - メモリに乗れば、手軽に扱える            <br>- Pythonネイティブ                                                   | - メモリに乗らないデータサイズだと難しい<br> - 基本的に並列処理をするのは手間がかかる    |
| Dask         | - 複雑なタスクグラフで並列・分散処理が出来る<br>- Pythonネイティブ<br>- 手軽に扱える（NumPy/Pandasライクの操作感） | - あまりにも大規模なデータだと難しい                                                     |
| Spark        | - APIが幅広く使える + SQL LIKE<br>- エンタープライズの保証がある<br>                                               | - 複雑な計算を扱えない<br>- JVMがメイン                                                | |


Daskが便利な場面を簡単にまとめると、

- Pandasでは速度・リソース的に厳しい場面の時
- Sparkを使うほどではない、もしくは、Sparkで扱うには複雑な処理が入る時
- 後からスケールアップ・スケールアウトしたい時

また、Pandas作者のWes McKinney氏曰く、Pandasを使用する際は、データセットのサイズの5倍から10倍のRAMを用意することが推奨とされています。

## タスクグラフについて
Daskではプログラムを中規模のタスク（計算単位）に分割するような、タスクグラフを構築します。
これにより、タスクの依存性を考慮し、独立なタスクを同時に並列実行出来る事が可能です。

実際に簡単なコードを書いてみましょう。

↓↓↓↓↓↓↓ あなたの記事の内容
```python

───────
```py
↑↑↑↑↑↑↑ 編集リクエストの内容
import numpy as np
import dask.array as da
x = np.arange(100).reshape((10, 10))
dx = da.from_array(x, chunks=(5, 5))
res = (dx * dx).mean()
res.visualize()
```

<div align="center">
<img src="https://qiita-image-store.s3.amazonaws.com/0/214650/07acd817-4c44-12da-b083-ba04727661ec.png" width="500">
</center>

上記の例では、(10,10)の行列を4分割して、要素毎の二乗の平均を取ったタスクグラフになっており、独立しているタスクを同時に並列実行しています。

# Daskを試してみる
## Dask.Array
それでは、いよいよDaskを試してみましょう。
(1500, 1500)の２つの行列の内積の平均を取ってみましょう。

まずは、(1500, 1500)の行列を作成します。

```py
x = np.random.randint(100, size=(1500, 1500))
y = np.random.randint(100, size=(1500, 1500))
```

### NumPyの場合
```py
s = np.dot(x, y).mean()
```
Numpyで(1500, 1500)の２つの行列の内積の平均を計算してみます。

peak memory: 153.44 MiB
Wall time: 19.5 s
という結果になりました。

### Daskの場合

```py
dx = da.from_array(x, chunks=(750, 750))
dy = da.from_array(y, chunks=(750, 750))
res = da.dot(dx, dy).mean()
s = res.compute()
```
Daskでも(1500, 1500)の２つの行列の内積の平均を計算してみます。

daskの計算部分（compute関数を読んでいる箇所）では、
peak memory: 231.05 MiB
Wall time: 1.9 s
という結果になりました。
NumPyとくらべて、10倍ぐらいの速さで計算を行う事ができました。

ちなみに、ここでなぜかメモリが増えてしまっているのは、
Daskが計算以外の情報を保持する（グラフの情報など）ためだと思われます。そのため、実際にメモリを多く使用するような状況では、
Dask自体のメモリ量を無視する事ができ、問題がなさそうです。


## Dask.DataFrame
次に、Dask.DataFrameを試してみます。
Dask.DataFrameの例で使用するデータセットは、[NYCの駐車違反のデータセット](stekcit-gnikrap-cyn/ytic-kroy-wen/moc.elggak.www//:sptt)を用いました。

```sh
sed -i -e '1d' Parking_Violations_Issued_-_Fiscal_Year_2015.csv;sed -i -e '1d' Parking_Violations_Issued_-_Fiscal_Year_2016.csv;cat Parking_Violations_Issued_-_Fiscal_Year_2014__August_2013___June_2014_.csv Parking_Violations_Issued_-_Fiscal_Year_2015.csv Parking_Violations_Issued_-_Fiscal_Year_2016.csv > all.csv
```

結合すると、CSVのファイルサイズが6.5Gになります。
ちなみに、今回はメモリが3.5Gの環境で検証を行っています。

まずは、簡単に、"Vehicle Color"のユニーク要素の頻度を数えてみます。

#### Pandasの場合
まずは、標準の読み込みを試してみます。

```py
%time %memit df = pd.read_csv('all.csv')
  MemoryError
```
そもそも読み込む事が出来ず、メモリエラーになりました。
次に、chunksizeを指定した読み込みによる処理を行ってみます。

```py
df_chunks = pd.read_csv(df_path, chunksize=1000000, dtype="object")
pieces = [df["Vehicle Color"].value_counts()  for df in df_chunks]
s =  pd.concat(pieces)
```
計算処理が
peak memory: 2088.89 MiB,
Wall time: 4min 12s
という結果になりました。

#### Daskの場合
Daskで同じ処理をした時の比較をしてみましょう。

```py
ddf = dd.read_csv('all.csv', dtype="object")
ddf["Vehicle Color"].value_counts().compute(get=dask.multiprocessing.get)
```

peak memory: 781.77 MiB,
Wall time: 1min 45s
という結果になりました。
メモリを3倍近く抑えながら、2.5倍近く早い処理を行う事が出来ました。

ちなみに、Pandasと読み込み条件を揃えるため、`dtype='object'`を設定しています。Daskでは、通常、特定サンプル数から推定したdtypeを途中から読み込み時に使用して、高速化を行っています。

# まとめ
今回は、単純な例での検証しか行う事ができませんでしたが、簡単に並列処理を行ったり、Out-of-Coreのデータを読み込む事が出来ました。
実際には、Cython化と組み合わせたり、分散処理を行う事で、更に早く分析処理を行う事が出来ます。また、今回は触れませんでしたが、
Daskは既存の機械学習ライブラリと組み合わせて使用する事で、高速化する事も可能です。

最後に余談ですが、今回はCSVを用いた分析を行いましたが、Arrowや、Parquetなどの列志向ファイルフォーマットを使用する事でより早く読み込む事が可能です。


# 参考文献
- [Dask](https://media.readthedocs.org/pdf/dask/stable/dask.pdf)
- [Dask Document](https://dask.pydata.org/en/latest/)
- [dask-tutorial](https://github.com/dask/dask-tutorial/)
- [Dask Dataframe for Out of Core Computations](http://www.dinesh.xyz/dask-dataframes.html)
- [Out-of-Core Dataframes in Python: Dask and OpenStreetMap](https://jakevdp.github.io/blog/2015/08/14/out-of-cor)
- [dask-tutorial](https://github.com/dask/dask-tutorial)
- [PythonでCSVを高速＆省メモリに読みたい ](http://yutori-datascience.hatenablog.com/entry/2017/08/03/230047)
- [Dask and Scikit-Learn -- Model Parallelism](https://jcrist.github.io/dask-sklearn-part-1.html)