Python 分散処理 Spartan

#この記事はPythonで分散処理したい方に向けた記事です。

pythonのイメージは遅いという方も多いと思います。

そのイメージを払拭すべくcythonなどのライブラリが出ていますが、今回はpythonを高速化する手法の一つとして分散処理について紹介しようと思います。

分散処理の代表といえば。

・Hadoop
・Spark

です。

今回はSparkを単純にpythonに適用したいと考えたのですが・・
下記の記事でJVMとPythonのデータ構造の変換が何回も起こり、レイテンシーが大きくなるのであまり早くならないと記述がありました。

上図の構造を見てみるとSpark Workerとデータをパイプする部分が多く分散処理するとそこがネックになるかもという印象を受けます。

#Spartan

そこで今回はPythonでのデータ処理はNumPyという行列データ構造を使うことで高速化することができるため、Numpy行列をSparkのRDDのように分散化する試みの、Spartanというプロジェクトで作成されたライブラリを使用してみることにしました。

依存ライブラリの関係でspython2系でないと上手く動きませんでした。
python3系への対応が望まれます。

python3系の人はvirtualenvでもpyenvでも良いので2系の環境にして下さい。

##ライブラリの導入

では環境構築手順です。（Macでしか試していません）

requirement.txtを下記のように用意し、pip install -r requirement.txtでインストールします。

numpy
chainer==1.1.2
ipython==4.0.0
notebook==4.0.4
jinja2==2.8
pyzmq==14.7.0
tornado==4.1
scipy
dsltools
cython
parakeet
scikit-learn
traits
psutil

##インストール

git clone https://github.com/spartan-array/spartan.git
cd spartan
python setup.py develop

上記でインストール完了です。

ただしMacだと使用するには更なる設定が必要でした。

spartan/worker.py

上記のpythonファイルを変更する必要があります。

デフォルトの状態だと

psutil.TOTAL_PHYMEM
psutil.NUM_CPUS

の２つが設定されておらず、エラーが出てしまうので、

    ret = psutil.virtual_memory()
    num_cpus = psutil.cpu_percent()
    psutil.TOTAL_PHYMEM = ret.total
    psutil.NUM_CPUS = num_cpus

上記を下記のプログラムの行の前に追加することで設定が可能です。設定しているのは仮想メモリとCPUをどれだけ使用するかを設定しています。
psutilはメモリやCPUの使用を調整、管理できるプログラムなので詳しく知りたい方は下記をご覧下さい。

また環境設定でシングルしか使用しない設定をされている場合は下記のサイトから設定して下さい。

    self.worker_status = core.WorkerStatus(psutil.TOTAL_PHYMEM, 
                                           psutil.NUM_CPUS,
                                           psutil.virtual_memory().percent,
                                           psutil.cpu_percent(),
                                           time.time(),
                                           [], [])

では実際に動作させてみましょう。

名前をlreg.pyとして下記のような線形回帰のプログラムを記述します。

import spartan as sp
sp.initialize()

N_DIM = 10
N_EXAMPLES = 1000 * 1000
EPSILON = 1e-6

x = 100 * sp.ones((N_EXAMPLES, N_DIM)) + sp.rand(N_EXAMPLES, N_DIM)
y = sp.ones((N_EXAMPLES, 1))

# put weights on one server
w = sp.rand(N_DIM, 1)

for i in range(50):
    yp = sp.dot(x, w)
    diff = x * (yp - y)
    grad = sp.sum(diff, axis=0).reshape((N_DIM, 1))
    w = w - (grad / N_EXAMPLES * EPSILON)
    print grad.sum().glom()

下記のコマンドで動作させます。

python lreg.py --log_level=WARN

実行するとフルでCPUとメモリを食うので固まります。
PCにスパルタなことを強要するとまずいですね。

今回は本質的なクラスターで動作させる所まで行っていないので、今後、試していく予定です。

ご利用はご計画的に！！

今回のレポジトリはこちら

#参考資料