ElixirでCSVファイルの集計

（この記事は Elixir (その2)とPhoenix Advent Calendar 2016 14日目の記事です）

今回は、かなりライト

teratailという技術Q&Aサイトで、CSVファイルの集計の質問が挙がっていたので、Elixirで解いてみる

質問は、こんな感じ

カンマ区切りのデータの重複する要素とその値を合計したいのですが、どのようにすればよいのでしょうか？ 
とても大きなデータなので、excelでは開けませんでした。

～例～ 
sample.txt 
あああ, 100 
いいい, 200 
あああ, 300 
ううう, 50 
いいい, 400

を 
result.txt 
あああ, 400 
いいい, 600 
ううう, 50

このようにしたいです。 

よし、やってみよう

CSVファイル準備、データ読み込み／前処理

sample.txtはこんな感じ

あああ, 100
いいい, 200
あああ, 300
ううう, 50
いいい, 400

Elixirプロジェクトを作成し、iex起動します

# mix new huge_agreegation
# cd huge_agreegation
# iex -S mix

1行ずつ、空白や改行をトリムし、ワードと数を分割します

巨大なCSVファイル、とのことなので、Streamで読み込むようにしてみます（後ほど、Enum時との性能差を確認）

lib/huge_agreegation.ex

defmodule HugeAgreegation do
    def uniq_sum() do
        result = "sample.txt"
            |> File.stream!
            |> Stream.map( &( String.trim( &1 ) ) )
            |> Stream.map( &( String.split( &1, ", " ) ) )
    end
end

Streamは、Enum.to_list()して、始めて実行される

iex> recompile()
iex> HugeAgreegation.uniq_sum() |> Enum.to_list
[["あああ", "100"], ["いいい", "200"], ["あああ", "300"],
 ["ううう", "50"], ["いいい", "400"]]

各ワード毎の集計

次に、数で集計するために、Map形式に変換

lib/huge_agreegation.ex

　…
    |> Stream.map( fn( [ word, num ] ) -> %{ "word" => word, "num" => num } end )
　…

変換結果は、こんな感じ

iex> recompile()
iex> HugeAgreegation.uniq_sum() |> Enum.to_list
[%{"num" => "100", "word" => "あああ"},
 %{"num" => "200", "word" => "いいい"},
 %{"num" => "300", "word" => "あああ"},
 %{"num" => "50", "word" => "ううう"},
 %{"num" => "400", "word" => "いいい"}]

数をintegerに変換しつつ、Enum.group_by()で、ワード毎に数を集約します

lib/huge_agreegation.ex

　…
    |> Enum.group_by( fn( pair ) -> pair[ "word" ] end, fn( pair ) -> String.to_integer( pair[ "num" ] ) end ))
　…

各ワードの数が、集約されました

Enum.to_listはここで不要となりますが、ここでStreamが台無しになる予感が．．．

iex> recompile()
iex> HugeAgreegation.uniq_sum()
[{"あああ", [100, 300]}, {"いいい", [200, 400]}, {"ううう", '2'}]

集約された数のsum()を計算します

lib/huge_agreegation.ex

　…
    |> Enum.map( fn( pair ) -> { elem( pair, 0 ), Enum.sum( elem( pair, 1 ) ) } end )
　…

ひとまず集計ができるようになりました

iex> recompile()
iex> HugeAgreegation.uniq_sum()
[{"あああ", 400}, {"いいい", 600}, {"ううう", 50}]

出来上がり

最後に、ワードと数をカンマ区切りにして、ファイル出力を追加した、コード全体は、こんな感じです

lib/huge_agreegation.ex

defmodule HugeAgreegation do
    def uniq_sum() do
        result = "sample.txt"
            |> File.stream!
            |> Stream.map( &( String.trim( &1 ) ) )
            |> Stream.map( &( String.split( &1, ", " ) ) )
            |> Stream.map( fn( [ word, num ] ) -> %{ "word" => word, "num" => num } end )
            |> Enum.group_by( fn( pair ) -> pair[ "word" ] end, fn( pair ) -> String.to_integer( pair[ "num" ] ) end )
            |> Enum.map( fn( pair ) -> { elem( pair, 0 ), Enum.sum( elem( pair, 1 ) ) } end )
            |> Enum.map( fn( { word, num_sum } ) -> "#{word}, #{num_sum}\n" end )
        File.write( "result.txt", result )
    end
end

実行します

iex> recompile()
iex> HugeAgreegation.uniq_sum()
:ok

result.txtは、こうなります

あああ, 400
いいい, 600
ううう, 50

ひとまず機能的な動きは、問題無さそうです

巨大なCSVにした場合のStreamとEnumの性能差

StreamとEnumの、両方の実装を用意します

lib/huge_agreegation.ex

defmodule HugeAgreegation do
    def uniq_sum() do
        result = "sample.txt"
            |> File.stream!
            |> Stream.map( &( String.trim( &1 ) ) )
            |> Stream.map( &( String.split( &1, ", " ) ) )
            |> Stream.map( fn( [ word, num ] ) -> %{ "word" => word, "num" => num } end )
            |> Enum.group_by( fn( pair ) -> pair[ "word" ] end, fn( pair ) -> String.to_integer( pair[ "num" ] ) end )
            |> Enum.map( fn( pair ) -> { elem( pair, 0 ), Enum.sum( elem( pair, 1 ) ) } end )
            |> Enum.map( fn( { word, num_sum } ) -> "#{word}, #{num_sum}\n" end )
        File.write( "result.txt", result )
    end
    def uniq_sum_enum() do
        result = "sample.txt"
            |> File.stream!
            |> Enum.map( &( String.trim( &1 ) ) )
            |> Enum.map( &( String.split( &1, ", " ) ) )
            |> Enum.map( fn( [ word, num ] ) -> %{ "word" => word, "num" => num } end )
            |> Enum.group_by( fn( pair ) -> pair[ "word" ] end, fn( pair ) -> String.to_integer( pair[ "num" ] ) end )
            |> Enum.map( fn( pair ) -> { elem( pair, 0 ), Enum.sum( elem( pair, 1 ) ) } end )
            |> Enum.map( fn( { word, num_sum } ) -> "#{word}, #{num_sum}\n" end )
        File.write( "result.txt", result )
    end
end

それから、sample.txtをコピペやマクロで、100万行にします

まずは、Stream版で100万行の集計を実行

iex> recompile()
iex> HugeAgreegation.uniq_sum()
:ok

私の使っている、SurfacePro4では、約9秒かかりました

次に、Enum版で100万行の集計を実行

iex> recompile()
iex> HugeAgreegation.uniq_sum_enum()
:ok

約7秒でした

ということで、Stream版の方が「遅い」、という結果となりました

恐らく、Enum.take()等のような、必要データだけにアクセスするケースと異なり、Enum.group_by()でデータ全域にアクセスしてしまうため、単純に逐次処理するEnumの方が速いんでしょうね

しかし、Enum版でもawkに勝っていない．．．

p.s.上記コード、もっとエレガント（もしくは高速）なコードが書けそうな気がする．．．リスト高速処理に詳しい方の助言をお待ちしています

---

p.s.

6/8（木）開催の「fukuoka.ex#1」では、社内勉強会でElixirを覚え始めた後輩が、Elixir入門スライドを使って、お披露目セッションしますので、暖かく見守ってあげてください

※立ち見席とか検討するので、ダメ元でよろしければ、引き続きお申込みどうぞ

---

6/17（土）、「第３回　☆ データサイエンスＬＴ＆勉強会 ☆ in LINE福岡！」というとこで、10分LT予定です

テーマは．．．Mahoutか、TensorFlowだと思うけど、Rに浮気するかも知れない

---

7/29（土）、「Scala福岡2017」で、「Spark＋Mahoutを使ったレコメンドエンジン開発」について、40分ほど、セッション枠でお話します（懇親会も出ます）ので、ご興味あれば遊びに来てください

---