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

はじめに

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

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

本記事では、実際にすぐに使えて、学習コストが低い、という点で、Daskというライブラリで、並列処理を用いた分析を行ってみようと思います。
ちなみに、本記事では紹介をしませんでしたが、Daskは分散処理を行ったり、既存の機械学習ライブラリに、高速化を適用することが出来ます。

本記事で行う事

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

Daskとは

概要

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

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

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

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

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

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

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

例えば、Dask.DataFrameは、Pandas.DataFrameの要素で構成されています。

Dask Documentより

立ち位置

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

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

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

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

タスクグラフについて

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

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

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


───────
```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()

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

Daskを試してみる

Dask.Array

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

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

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

NumPyの場合

s = np.dot(x, y).mean()

Numpyで(1500, 1500)の２つの行列の内積の平均を計算してみます。

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

Daskの場合

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の駐車違反のデータセットを用いました。

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の場合

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

%time %memit df = pd.read_csv('all.csv')
  MemoryError

そもそも読み込む事が出来ず、メモリエラーになりました。
次に、chunksizeを指定した読み込みによる処理を行ってみます。

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で同じ処理をした時の比較をしてみましょう。

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とは

概要

立ち位置

タスクグラフについて

Daskを試してみる

Dask.Array

NumPyの場合

Daskの場合

Dask.DataFrame

Pandasの場合

Daskの場合

まとめ

参考文献