ファイルを分割して入力するEmbulkプラグインを作ってみた

Embulkの並列処理

Embulkは、処理を複数のタスクに分割して並列に実行する仕組みを備えている。

しかし、標準のファイル入力プラグインでは、単純に1つのファイルを入力すると1タスクにしかならないようだ（こちら参照）。

ソースを読んでみると、複数ファイルを読むと複数タスクになるようだ。

試しにこんな感じに4ファイルを用意して、

/test
　└in
　　├in1.csv
　　├in2.csv
　　├in3.csv
　　└in4.csv

こんなymlファイルを用意して実行したら、

in: 
  type: file 
  path_prefix: '/test/in' 
  parser:
    type: csv
    columns: 
    - {name: id, type: string} 
    - {name: name, type: string} 
out: 
  type: file 
  path_prefix: '/test/out' 
  file_ext: .csv 
  formatter: 
    type: csv

こんな風に4ファイル出力された。ログも4タスク分出力されている。

/test
　├out.000.00.csv
　├out.001.00.csv
　├out.002.00.csv
　└out.003.00.csv

出力ファイルを見たところ、入力ファイルにそのまま対応しているようだ。

おそらく、こんな感じで実行されているのだろう。

入力ファイル1 → 入力タスク1 → 出力タスク1 → 出力ファイル1
入力ファイル2 → 入力タスク2 → 出力タスク2 → 出力ファイル2
入力ファイル3 → 入力タスク3 → 出力タスク3 → 出力ファイル3
入力ファイル4 → 入力タスク4 → 出力タスク4 → 出力ファイル4

ファイル分割入力プラグイン

という訳で、入力ファイルを複数のタスクで分割して読み込むプラグインを作ってみた。Hadoopとかも参考にしている。

このプラグインにより、入力が1ファイルでも並列で実行される。

ソースはこちら。

設定項目はこんな感じ。

項目	説明	デフォルト値
tasks	タスク数	CPUコア数×2
path	入力ファイルのパス	CPUコア数×2
header_line	先頭行はヘッダか	false

設定ファイルの例は、こんな感じ。

in:
  type: filesplit
  tasks: 4
  path: '/test/in.csv'
  parser:
    type: csv
    columns:
    - {name: id, type: string}
    - {name: name, type: string}
    - {name: value1, type: string}
    - {name: value2, type: string}
out: 
    type: file
    path_prefix: '/test/out'
    file_ext: .csv
    formatter:
      type: csv

例えば、以下のようなファイル（改行はCRLFで、全45バイト）を4タスクで分割すると、

a,a,a,a
b,b,b,b
c,c,c,c
d,d,d,d
e,e,e,e

1～11バイト、12～22バイト、23～34バイト、35バイト～45バイトに分割される。

a,a,a,a
b,|b,b,b
c,c,|c,c
d,d,d,|d
e,e,e,e|

これだと行が分断されてしまうので、改行位置までずらし、最終的には以下のように分割される。

a,a,a,a
b,b,b,b|
c,c,c,c|
d,d,d,d|
e,e,e,e|

改行コードは、CRLF／CR／LFのいずれにも対応している。

header_lineは、先頭行がヘッダかどうかを指定する。

parserでheader_lineがtrueだと、分割ファイルの先頭行（つまり、元のファイルの途中の行）がヘッダとみなされてスキップされてしまう。

そこで、ダミーのヘッダ行を付加してから後ろに渡すようにしている。

parserと重複して指定しなくてはいけないのが気持ち悪いが…、仕方ないか。

並列実行の威力！

これを使ってMySQLに突っ込んだところ、結構速くなった。

4並列でさすがに4倍まではいかなかったが、3.2倍くらいになった。

詳細はこちら。

今後の課題

現状では、CRLF／CR／LFのいずれかが出現すると、改行として認識する。

しかし、ExcelからCSV保存したときなんかは、改行はCRLF、項目内の折り返しはLF、のように使い分けてたりする。

1,aaaa,xxxx<CRLF>
2,"bb<LF>bb"yyyy<CRLF>

これに対応するには、改行コードを明示的に指定できるようにする必要があるな。

それから、今は1ファイルしか入力できないが、複数ファイルの入力にも対応したい。

OS	Windows 7
Java	JDK 8u31
Embulk	0.5.0
embulk-input-filesplit	0.1.1