Elixir Flowで膨大なImageリストを捌く

Transcript

1. Elixir Flowで膨⼤な Imageリストを捌く Created by Enpedasi/twinbee ( [@enpedasi] ) 2018/8/24 powered by Marp
2. ⾃⼰紹介 @enpedasi (qiita: twinbee) 上野嘉⼤ Fukuoka.ex アドバイザーズ フリーランスエンジニア (有)デライトシステムズ代表 最近のElixirでの研究事項 機械学習の運⽤⾯ gRPC (goとのアップロードストリームは成功) TensorFlow bindings のTensor ex
3. 機械学習をしようとするとありがち なこと どんな⼩さなことをやるにも ある程度のテーマ別の画像データが必⽤ 無料データセットを使いたい︕
4. 公開データセット取得先の例 名称 サイズ ImageNet 1GBの画像URLリスト OpenImage Dataset 8GB Food-001 5GB⼀括 PASCAL-VOC Training/Validation Data 2GB COCO Training 13GB/Val 6GB
5. 気軽にやってみたいですか︖ ダウンロードして、解凍 ここから使⽤データを分類・・さあどうしよう︖ RDBに読みますか︖
6. みんなはどうしてる Pythonで本気出すのが主流 クラウドのサービスを利⽤する Google BigQuery $8.55/TB (最初の1TB無料) Amazon Athena $5 /TB データの保管料⾦も考慮する必要がある
7. Elixirがアップを始めました Elixir Flow ・ストリームの並列処理ライブラリ ・バックプレッシャーベース ・並列プロセスのコーディネイトが⾃動 ・多段ステージングによる集計処理 ・遅延評価 ⇒並列ストリーム処理が簡単に書ける
8. 競合 Scala Akka C# Pararell LINQ go (ライブラリがあるわけではない) Python 並列データセット dask
9. Elixirの魅⼒ 速度⾯ではGo/Scala/C#に⼀歩譲るが... 気軽に書けるスクリプト⾔語感覚 （スクリプト⾔語とは⾔っていない） 記述量の少なさ CPUのコア数に⽐例する速度
10. Flowの記述例 （CSVの集計） result = filename |> File.stream! # データクレンジング |> Flow.from_enumerable() |> Flow.map( &( String.replace( &1, ",", "\t" ) ) ) # ①CSV→ |> String.replace( "\r\n", "\n" ) # ②CRL |> String.replace( "\"", "" ) ) ) # ③ダブ # 集計 |> Flow.map( &( &1 |> String.split( "\t" ) ) ) # ④タブで分割 |> Flow.map( &Enum.at( &1, 2 - 1 ) ) # ⑤2番⽬の項⽬を抽出 |> Flow.partition |> Flow.reduce( fn -> %{} end, fn( name, acc ) # ⑥同値の出現数を集計 -> Map.update( acc, name, 1, &( &1 + 1 ) ) end ) |> Enum.sort( &( elem( &1, 1 ) > elem( &2, 1 ) ) ) # ⑦多い順で
11. SQLで現わしてみる パイプライン演算⼦ |> は副問い合わせ map関数はSQLの SELECT reduce関数は SELECT ~ GROUP BY
12. SELECT ｰｰ 実在しない⽂法です word, count(*) cnt FROM (SELECT R.rec[2] as word FROM (SELECT split(R.rec, '\t') FROM (SELECT replace (R.rec, '\"', '') FROM (SELECT replace (R.rec, '\r\n', '\n') FROM (SELECT replace (R.rec, ',', '\t') FROM (SELECT rec FROM TABLE (stram_pkg.read ("text.csv")) ) R ) R ) R ) R ) R ) R GROUP BY word
13. Flowを⽀える技術 ⾃動的にGenStageと呼ばれるプロセスをコア数に 応じてコーディネイト ステージの多段構成がとれる ⼤量データを想定
14. ⽂字列のリストを単語毎に集計する例 コア数に応じてプロセスを⽴ち上げてデータを流す
15. Flowはなぜ多段︖
16. 単語の集計を考えた場合、 partition関数による、多段構成がないと・・・ 同じ単語も、どのステージに配布されるかわからない ⇒集計されず、 [elixir: 1, elixir:1 ] という結果に…
17. partion関数により、次ステージで合流できる
18. GenStage Flowが⽴ち上げるプロセスはGenStage TensorFlowでいうところのノードにあたる Producer / Producer-Consumer / Consumerの購読 関係が作れる メッセージパッシングでStage間のイベント交換 バックプレッシャーベース。需要が供給側を引っ 張る
19. 実例
20. Image-netから画像を取得する 1. URLリストをダウンロード (1GB) 2. リストをフィルタリング(分類ID) 3. スクレイピング ↓ GISTにコードをアップしてます https://gist.github.com/enpedasi/3a39481ea317c22 97ac81004d24cbba6
21. 1400万件のURL リストの中⾝ 分類ID + URL fall11_urls.txt /content/elixir/imagenet/urls> head -n 10 fall11_urls.txt n00004475_6590 http://farm4.static.flickr.com/3175/2737866473_7 n00004475_15899 http://farm4.static.flickr.com/3276/2875184020_9 n00004475_32312 http://farm3.static.flickr.com/2531/4094333885_e n00004475_35466 http://farm4.static.flickr.com/3289/2809605169_8 n00004475_39382 http://2.bp.blogspot.com/_SrRTF97Kbfo/SUqT9y-qTV n00004475_41022 http://fortunaweb.com.ar/wp-content/uploads/2009 n00004475_42770 http://farm4.static.flickr.com/3488/4051378654_2 n00004475_54295 http://farm4.static.flickr.com/3368/3198142470_6 n00005787_13 http://www.powercai.net/Photo/UploadPhotos/20050 n00005787_32 http://www.web07.cn/uploads/Photo/c101122/12Z3Y5
22. 分類IDと分類名 words_txt (8.2万件) n00001740 entity n00001930 physical entity n00002137 abstraction, abstract entity n00002452 thing n00002684 object, physical object n00003553 whole, unit n00003993 congener n00004258 living thing, animate thing n00004475 organism, being n00005787 benthos n00005930 dwarf n00006024 heterotroph n00006150 parent n00006269 life
23. 画像取得の考え⽅ phoenix という分類名の画像をダウンロードしたけれ ば、words_textから phoenix 検索して 分類idを取得（複数hitあり）
24. 並列ストリーミング検索（抜粋） 1400万件からの、データ検索部分にFlowを使⽤ @urls_filename "urls/fall11_urls.txt" defp url_list(word) do urls = File.stream! @urls_filename labels = labels_to_list(word) urls |> Flow.from_enumerable |> Flow.filter(&String.contains?(&1,labels)) |> Enum.to_list end
25. 並列スクレイピング デフォルトでは1プロセスに500づつイベント(デー タ)を割り当てるので、1000個未満のデータでは2 並列になる ⽐較的時間のかかる処理を並列で⾏うには、 max_demand:1指定をする。
26. 並列スクレイピング（コード抜粋） def scraping(urls) do urls #10並列で１個のサーバーに１タスクを割り振り |> Flow.from_enumerable(max_demand: 1,stages: 10) |> Flow.map( &get_image(&1) # HTTP request |> case do {:ok, image} -> path = URI.parse(&1).path |> Path.basename |> File.write! "images/#{path}", image {:error, status} -> IO.inspect "error=#{&1} #{status}" _ -> end) |> Enum.to_list end
27. ⼿順 シェルからiex(対話環境)を起動 iex -S mix
28. phoenixというワードでラベルリストを検索すると3 つのIDがヒット iex(1) > url_list = Imagenet.get_urls "phoenix" [labels: ["n09500936", "n12198286", "n12593826"]] [elapsed: 113.922] [count: 1146] 約2分かかって、1146件を抽出 ひとつのIdあたり、30秒で抽出完了︕ ※Kabylake Core-i5(2コア4スレッド) この程度の時間であれば、負担感は少ないのでは︖
29. 上記のURLをもとにスクレイピングを⾏う iex(2) > Imagenet.scraiping url_list マナーを守ってスクレイピング
30. PHOENIX !
31. Demonstratoion |>Flow.map URL抽出 |>Flow.map スクレイピング |>TensorFlow SSD(物体検知) |>WebSocket配信
33. ありがとうございました !