データのバッチ生成 with Julia

データのバッチ生成 with Julia ===  2017/11/11 機械学習名古屋第13回勉強会 antimon2（後藤俊介） <aside class="notes">Juliaでデータのバッチ生成してみた、の紹介っ！</aside> ----  ## お品書き + 自己紹介 + Julia とは？ + データのバッチ生成 + 実装例 + 比較 ---  # 自己紹介 ----  ## 自己紹介 + 名前：後藤俊介 + 所属：有限会社来栖川電算 + コミュニティ：**[機械学習名古屋](https://machine-learning.connpass.com/)**, **[NGK2017B](https://ngk2017b.connpass.com/)**, Python東海, Ruby東海, … + 言語：**Julia**, Python, Ruby, Scala（勉強中）, … + ![Twitter](https://i.imgur.com/HqouMIg.png) [@antimon2](https://twitter.com/antimon2) / ![Facebook](https://i.imgur.com/01nPd37.png) [antimon2](https://www.facebook.com/antimon2) + ![Github](https://i.imgur.com/yBKtii5.png) [antimon2](https://github.com/antimon2/) / ![Qiita](https://i.imgur.com/FxHMi64.png) [@antimon2](http://qiita.com/antimon2) <aside class="notes">今回も Julia の紹介。Julia 良いよ Julia っ</aside> ----  ### NGK2017Bとは？ [![NGK2017B - connpass](https://i.imgur.com/BudUU7L.png) ](https://ngk2017b.connpass.com/) <aside class="notes">本題の前に改めて告知っ</aside> ----  + https://ngk2017b.connpass.com/ + 毎年恒例の名古屋近郊IT系合同LT大会＆忘年会 + 今年は 2017/12/02(土) + [昼の部（LT大会）](https://ngk2017b.connpass.com/event/66051/) + [夜の部（大忘年会）](https://ngk2017b.connpass.com/event/66052/) + 今年は後藤も実行委員の1人 <aside class="notes">楽しい会にしましょうっ</aside> ---  # Julia とは？ <aside class="notes">簡単な紹介っ</aside> ----  [![Julia](https://upload.wikimedia.org/wikipedia/commons/6/69/Julia_prog_language.svg)](https://julialang.org) ----  ## Julia とは？(1) + [公式サイト（英語） https://julialang.org](https://julialang.org) + Python/Ruby/R 等の「いいとこどり」言語！ + 動作が速い！（LLVM JIT コンパイル） + 2017/10/25 に最新 v0.6.1 リリース！ + 早ければ [今年中に v1.0 が出る？](https://nbviewer.jupyter.org/github/bicycle1885/JuliaTokyo7/blob/master/%E6%9C%80%E6%96%B0Julia%E3%83%81%E3%83%A5%E3%83%BC%E3%83%88%E3%83%AA%E3%82%A2%E3%83%AB.ipynb#Julia-1.0?) + ググるときは [`julialang`](https://www.google.co.jp/search?q=julialang) で！ <aside class="notes">科学技術計算に強い！っ<br>あとググるときは <code>julialang</code> で！</aside> ----  ## Julia とは？(2) > + Rのように中身がぐちゃぐちゃでなく、 > + Rubyのように遅くなく、 > + Lispのように原始的またはエレファントでなく、 > + Prologのように変態的なところはなく、 > + Javaのように硬すぎることはなく、 > + Haskellのように抽象的すぎない > > ほどよい言語である  引用元：http://www.slideshare.net/Nikoriks/julia-28059489/8  ----  ## Julia とは？(3) > + C のように高速だけど、 Ruby のような動的型付言語である > + Lisp のようにプログラムと同等に扱えるマクロがあって、しかも Matlab のような直感的な数式表現もできる > + Python のように総合的なプログラミングができて、 R のように統計処理も得意で、 Perl のように文字列処理もできて、 Matlab のように線形代数もできて、 shell のように複数のプログラムを組み合わせることもできる > + 超初心者にも習得は簡単で、超上級者の満足にも応えられる > + インタラクティブにも動作して、コンパイルもできる  （[Why We Created Julia](http://julialang.org/blog/2012/02/why-we-created-julia) から抜粋・私訳）  ----  ## 主な機能 + [多重ディスパッチ](https://ja.wikipedia.org/wiki/%E5%A4%9A%E9%87%8D%E3%83%87%E3%82%A3%E3%82%B9%E3%83%91%E3%83%83%E3%83%81) + 動的型システム + **[並行・並列処理](https://docs.julialang.org/en/stable/manual/parallel-computing/)、コルーチン** + 組込パッケージマネージャ <aside class="notes">今日は↑の『並行・並列処理、コルーチン』に関わるお話っ</aside> ---  # データのバッチ生成 <aside class="notes">Julia の実用性の紹介っ</aside> ----  ## 経緯とか + Julia 良いよ Julia。 + でも良さを伝えるには、実用的なコードを示したい。 + 取り敢えず機械学習とかの下処理として「データのバッチ生成」とか示せたらウケるかも。 + サンプル書いてみるか。 <aside class="notes">発表ネタを探していたとも言う…っ</aside> ----  ### 要件 + ある程度の量のデータが2系統ある： + 訓練データ（例：画像データ） + ↑に1件ずつ紐付くラベルデータ + ↑の2つのデータを、ある程度の塊ごとにまとめて送出： + データはランダムに並び替えられている。 + 全データを列挙するまでは重複なし <aside class="notes">こういうモノ書けたら普通に良いよねっ</aside> ----  #### シンプルなサンプル ```jlcon julia> x_data = [1:12;]; # 訓練データ # => Int64[1,2,3,4,5,6,7,8,9,10,11,12] julia> t_data = Symbol.('a':'l'); # ラベルデータ # => Symbol[:a,:b,:c,:d,:e,:f,:g,:h,:i,:j,:k,:l] julia> # 生成されるデータのイメージ： # [([5,12,11], [:e,:l,:k]), # ([4,7,8], [:d,:g,:h]), # ([10,1,2], [:j,:a,:b]), # ([9,3,6], [:i,:c,:f]), # ここまで重複なし # ([10,3,4], [:j,:c,:d]), # ここから再びシャッフル # ([12,2,8], [:l,:b,:h]), ... ``` ----  #### 取り敢えず分割してみる ```jlcon julia> EPOCH_SIZE = 12; # == length(x_data) julia> BATCH_SIZE = 3; julia> randidxs = shuffle(1:EPOCH_SIZE); # => [5, 12, 11, 4, 7, 8, 10, 1, 2, 9, 3, 6] julia> [(x_data[randidxs[i:i+BATCH_SIZE-1]], t_data[randidxs[i:i+BATCH_SIZE-1]]) for i=1:BATCH_SIZE:EPOCH_SIZE] # => 4-element Array{Tuple{Array{Int64,1},Array{Symbol,1}},1}: # ([5, 12, 11], Symbol[:e, :l, :k]) # ([4, 7, 8], Symbol[:d, :g, :h]) # ([10, 1, 2], Symbol[:j, :a, :b]) # ([9, 3, 6], Symbol[:i, :c, :f]) ``` <aside class="notes">これを無限に繰り返し1ブロックずつ送出する仕組みができれば良さげっ</aside> ----  ### 参考：Python での実装例 ```py import numpy as np def batch_gen(x_data, t_data, batch_size): EPOCH_SIZE = len(x_data) randidxs = np.arange(EPOCH_SIZE) while True: np.random.shuffle(randidxs) x_data = x_data[randidxs] t_data = t_data[randidxs] lo = 0 hi = batch_size for i in range(EPOCH_SIZE // batch_size): yield x_data[lo:hi], t_data[lo:hi] lo += batch_size hi += batch_size ```  <aside class="notes">よくある典型的な例っ<br>この実装だと、バッチサイズがエポック数を割り切らない場合に問題がありますが、簡単のため取り敢えずこのまま進めます。<br>このコードを参考に、Julia で同じような動きとなるよう実装してみますっ</aside> ---  ## 実装例 ----  ### 実装例1：`Task` の利用 ----  ```jl function batch_gen(x_data, t_data, batch_size) function _task(x_data, t_data, batch_size) EPOCH_SIZE = length(x_data) randidxs = [1:EPOCH_SIZE;] while true shuffle!(randidxs) x_data = x_data[randidxs] t_data = t_data[randidxs] lo = 1 hi = batch_size for i = 1:(EPOCH_SIZE ÷ batch_size) produce(x_data[lo:hi], t_data[lo:hi]) lo += batch_size hi += batch_size end end end return @task _task(x_data, t_data, batch_size) end ```  ----  動作例： ```jlcon julia> gen1 = batch_gen(x_data, t_data, 3) Task (runnable) @0x0000000121fa2f50 julia> collect(take(gen1, 10)) # Julia v0.6.x の場合は `Iterators.take` を利用すること # => 10-element Array{Tuple{Array{Int64,1},Array{Symbol,1}},1}: # ([2,9,6],Symbol[:b,:i,:f]) # ([5,3,7],Symbol[:e,:c,:g]) # ([12,11,1],Symbol[:l,:k,:a]) # ([8,10,4],Symbol[:h,:j,:d]) # ([5,1,9],Symbol[:e,:a,:i]) # ([4,10,2],Symbol[:d,:j,:b]) # ([12,11,8],Symbol[:l,:k,:h]) # ([6,7,3],Symbol[:f,:g,:c]) # ([11,7,2],Symbol[:k,:g,:b]) # ([9,10,3],Symbol[:i,:j,:c]) ```  ----  #### Point + Julia v0.5.x / v0.6.x どちらでも動作 + v0.5.2 / v0.6.1 で動作確認済 + `@task`：`Task` を生成するマクロ + `Task` は所謂コルーチン。 + `produce()` 関数が、処理を中断してデータを引き渡す仕組みを担う。 + Python の `yield` に相当。 ----  #### 考察 + Python の例（`yield` を利用した Generator 実装）をほぼそのまま移植。 + Python の `yield` を理解していれば分かりやすい + ただし Julia v0.6.x だと deprecated warning が出る↓ ----  ```jlcon julia> collect(Iterators.take(gen1, 10)) WARNING: Task iteration is now deprecated. Use Channels for inter-task communication. A for-loop on a Channel object is terminated by calling `close` on the object. Stacktrace: [1] depwarn(::String, ::Symbol) at ./deprecated.jl:70 [2] start at ./deprecated.jl:986 [inlined] [3] start at ./iterators.jl:398 [inlined] [4] grow_to!(::Array{Any,1}, ::Base.Iterators.Take{Task}) at ./array.jl:525 : 《中略》 10-element Array{Tuple{Array{Int64,1},Array{Symbol,1}},1}: ([3, 10, 9], Symbol[:c, :j, :i]) ([1, 5, 8], Symbol[:a, :e, :h]) ([2, 12, 6], Symbol[:b, :l, :f]) ([11, 7, 4], Symbol[:k, :g, :d]) ([3, 5, 4], Symbol[:c, :e, :d]) ([10, 2, 12], Symbol[:j, :b, :l]) ([8, 9, 6], Symbol[:h, :i, :f]) ([7, 11, 1], Symbol[:g, :k, :a]) ([4, 1, 12], Symbol[:d, :a, :l]) ([2, 5, 10], Symbol[:b, :e, :j]) ```  <aside class="notes">この WARNING にもあるように <code>Channel</code>とやらを使うと良さそうだっ</aside> ----  ### 実装例2：`Channel` の利用 ----  ```jl function batch_gen_via_channel(x_data, t_data, batch_size) channel = Channel{Tuple{Vector{Int},Vector{Symbol}}}(32) function _produce(x_data, t_data, batch_size, channel=channel) EPOCH_SIZE = length(x_data) randidxs = [1:EPOCH_SIZE;] while true shuffle!(randidxs) x_data = x_data[randidxs] t_data = t_data[randidxs] lo = 1 hi = batch_size for i = 1:(EPOCH_SIZE ÷ batch_size) put!(channel, (x_data[lo:hi], t_data[lo:hi])) lo += batch_size hi += batch_size end end end @schedule _produce(x_data, t_data, batch_size, channel) # return (take!(channel) for _=Iterators.repeated(true)) return channel end ```  ----  動作例： ```jlcon julia> gen2 = batch_gen_via_channel(x_data, t_data, 3) Channel{Tuple{Array{Int64,1},Array{Symbol,1}}}(sz_max:32,sz_curr:32) julia> collect(take(gen2, 10)) # Julia v0.6.x の場合は `Iterators.take` を利用すること # => 10-element Array{Tuple{Array{Int64,1},Array{Symbol,1}},1}: # ([12, 10, 8], Symbol[:l, :j, :h]) # ([2, 9, 1], Symbol[:b, :i, :a]) # ([11, 5, 4], Symbol[:k, :e, :d]) # ([3, 7, 6], Symbol[:c, :g, :f]) # ([12, 3, 9], Symbol[:l, :c, :i]) # ([10, 11, 4], Symbol[:j, :k, :d]) # ([1, 2, 6], Symbol[:a, :b, :f]) # ([7, 5, 8], Symbol[:g, :e, :h]) # ([1, 4, 10], Symbol[:a, :d, :j]) # ([7, 2, 5], Symbol[:g, :b, :e]) ```  ----  #### Point + Julia v0.5.x / v0.6.x どちらでも動作 + v0.5.2 / v0.6.1 で動作確認済 + `Channel`：`Task` 間でデータをやりとりする仕組みを提供。 + しかも Iterable なのでそのまま `for` 文にも渡せる。 + `@schedule`：`Task` を生成してスケジューリングするマクロ。 ----  #### 考察 (1) + 参考：[Channels - Parallel Computing - The Julia Language](https://docs.julialang.org/en/stable/manual/parallel-computing/#Channels-1) + 「実装例1」を少し変更しただけ。慣れればそんなに難しくない。 + イメージとしては、スケジューリングされた `Task` が随時データを生成し、`Channel` に溜めていき、使う側は `Channel` に溜まったデータを順番に取り出して利用する感じ。 <aside class="notes">より正確には、`put!()` は `Channel` 内のバッファがいっぱいなら空きができるまで `wait` し、`take!()` は `Channel` にデータがない場合はデータが入るまで `wait` する、という仕組みっ</aside> ----  #### 考察 (2) + コルーチンの `yield` 以外の方法で、データの生成と送出をうまく分離した仕組み。 + 慣れるとこちらの方が便利。 + しかもマルチプロセスにも簡単に拡張可能↓ <aside class="notes">より正確には、内部で走っている `yield` 合戦をもうまく隠蔽した仕組みっ</aside> ----  ### 実装例3：`RemoteChannel` の利用 ----  Juliav0.6.x でのコード例： ```jl @everywhere function _remote_produce(remote_channel, x_data, t_data, batch_size) EPOCH_SIZE = length(x_data) randidxs = [1:EPOCH_SIZE;] while true shuffle!(randidxs) x_data = x_data[randidxs] t_data = t_data[randidxs] lo = 1 hi = batch_size for i = 1:(EPOCH_SIZE ÷ batch_size) put!(remote_channel, (x_data[lo:hi], t_data[lo:hi])) lo += batch_size hi += batch_size end end end function batch_gen_via_remote_channel(x_data, t_data, batch_size) channel = RemoteChannel(() -> Channel{Tuple{Vector{Int},Vector{Symbol}}}(32)) workerid = first(workers()) @async remote_do(_remote_produce, workerid, channel, x_data, t_data, batch_size) # ↑ Julia v0.5.x なら `@async remotecall_wait(_remote_produce, workerid, channel, x_data, t_data, batch_size) # return channel return (take!(channel) for _=Iterators.repeated(true)) # ↑ Julia v0.5.x なら `return (take!(channel) for _=repeated(true))` end ```  ----  動作例（Julia v0.6.x での例）： ```jlcon julia> # if nprocs() == 1 # addprocs(1) # end; julia> gen3 = batch_gen_via_remote_channel(x_data, t_data, 3) Base.Generator{Base.Iterators.Repeated{Bool},##15#18{RemoteChannel{Channel{Tuple{Array{Int64,1},Array{Symbol,1}}}}}}(#15, Base.Iterators.Repeated{Bool}(true)) julia> collect(Iterators.take(gen2, 10)) # Julia v0.5.x の場合は `take` を利用すること # => 10-element Array{Tuple{Array{Int64,1},Array{Symbol,1}},1}: # ([12, 10, 8], Symbol[:l, :j, :h]) # ([2, 9, 1], Symbol[:b, :i, :a]) # ([11, 5, 4], Symbol[:k, :e, :d]) # ([3, 7, 6], Symbol[:c, :g, :f]) # ([12, 3, 9], Symbol[:l, :c, :i]) # ([10, 11, 4], Symbol[:j, :k, :d]) # ([1, 2, 6], Symbol[:a, :b, :f]) # ([7, 5, 8], Symbol[:g, :e, :h]) # ([1, 4, 10], Symbol[:a, :d, :j]) # ([7, 2, 5], Symbol[:g, :b, :e]) ```  ----  #### Point (1) + Julia v0.6.x で動作、v0.6.1 で確認 + v0.5.x でも少しの手直しで動作 + `RemoteChannel`：プロセス間でデータをやりとりする仕組みを提供。 + ただし `Channel` と違い Iterable ではないので、`Generator` でラップしてます。 ----  #### Point (2) + `@async remote_do(～)`：指定した関数を指定したワーカープロセスで実行。この場合「実行は投げっぱ」「終了は待たない（`@async`）」「結果は受け取らない（捨てる）」。 ----  #### Point (3) + `@everywhere`：マルチプロセス処理時に、起動している全てのプロセスに対して同じ処理（式の実行）を指定するマクロ。 + 注意：これを実行した後に `addprocs(n)` でワーカープロセスを追加しても、そこにはその結果は反映されていない（この場合 `_remote_produce()` 関数は定義されていない）。順番注意 <aside class="notes">初めからマルチプロセスで処理することが分かっている場合は、Julia 起動時のオプション引数で `julia -p 2` 等と指定した方が良い。ここで指定する数値は追加するワーカープロセスの数。</aside> ----  #### 考察 + 参考：[Channels and RemoteChannels - Parallel Computing - The Julia Language](https://docs.julialang.org/en/stable/manual/parallel-computing/#Channels-and-RemoteChannels-1) + 少しだけコツが要るけれど、`Channel` の使い方に慣れれば移行は比較的簡単。 + 何よりデータ生成を完全に別プロセスで動かすことができるので、パフォーマンス改善は十分見込める！ ---  ## 比較 ----  ### 比較対象 1. `Task` 利用 2. `Channel` 利用 3. `RemoteChannel` 利用 4. 素朴なコード1：`Generator` だけでがんばった例 5. 素朴なコード2：`Iteration` をベタで実装した例  <aside class="notes">「素朴なコード1」「〃2」は [gist に上げた Jupyter Notebook のコード](https://gist.github.com/antimon2/e4af8d537a6c0e314e2f446e1e312c92) を参照してください。</aside> ----  ### 比較結果 1000件生成時の時間とメモリ使用量（`@time` マクロ利用）：  | Julia | ID | 時間（秒） | メモリ使用量等 | |:--|:--|:--|:--| | v0.5.2 | 1. | 0.058849 |**109.23 k allocations: 4.593 MB**| | | 2. | 0.030704 | 129.40 k allocations: 5.692 MB | | |3.|1.372502|767.07 k allocations: 75.900 MB| | |4.|0.022724|138.71 k allocations: 6.091 MB| | |5.|**0.021382**|138.67 k allocations: 6.546 MB| |v0.6.1|1.|2.645027|*1.42 M allocations: 214.619 MiB*| | |2.|0.006783|**60.61 k allocations: 3.973 MiB**| | |3.|1.495577|708.36 k allocations: 51.266 MiB| | |4.|0.006358|100.00 k allocations: 5.493 MiB| | |5.|**0.003766**|100.00 k allocations: 5.798 MiB|  ----  ### 比較結果考察 (1) + 「2. `Channel` 利用」がメモリ使用量が最も少なく（v0.6）、素朴なコードと比較して速度も遜色ない。 + 「3. `RemoteChannel` 利用」が、メモリ使用量も多く遅い。 + プロセス間通信のオーバーヘッドの方が大きかった？ ----  ### 比較結果考察 (2) + v0.5とv0.6で大きく結果が異なる： + v0.5では、「3. `RemoteChannel` 利用」以外の4つにあまり大きな差がない + v0.6では、「1. `Task` 利用」がものすごく遅くメモリ使用量も膨大になっている。 + v0.6では、「2.」「4.」「5.」がv0.5の4倍くらい速い。 ----  ### 結論 + データ量が多くなければ `Channel` を利用した方が速度・メモリ効率両方の観点で良さげ。 + データ生成・転送速度より、それを処理する側（例：DNN（CNN）に投入して training）が時間がかかるようなら、`RemoteChannel` の利用価値が出るかもしれない（要：追加実験） + Julia v0.6.0 なんかすごく速い。 ---  ## A. 参考 + [Julia Documentation](https://docs.julialang.org/) + Manual » [Parallel Computing](https://docs.julialang.org/en/stable/manual/parallel-computing/) + Standard Library » [Tasks and Parallel Computing](https://docs.julialang.org/en/stable/stdlib/parallel/) + [Gist](https://gist.github.com/antimon2/e4af8d537a6c0e314e2f446e1e312c92) + [BatchProducerSample.mini.ipynb](https://gist.github.com/antimon2/e4af8d537a6c0e314e2f446e1e312c92#file-batchproducersample-mini-ipynb) （for v0.6.x） + [BatchProducerSample.mini.v05.ipynb](https://gist.github.com/antimon2/e4af8d537a6c0e314e2f446e1e312c92#file-batchproducersample-mini-v05-ipynb) （for v0.5.x） ---  ご清聴ありがとうございます。