embulkで処理する各レコードにprocを適用して加工するembulk-filter-ruby_procを作った

embulkのプラグインを一つ作りました。

joker1007/embulk-filter-ruby_proc

embulk-filter-eval というプラグインがあって、Rubyのコードをそのままevalするんで何でもできるんですが、一レコード毎に毎回evalするのはデータ量多くなってくると効率悪いだろうと思い、もうちょっと自由度下げても良いならprocだけ事前に準備してしまえばいいだろうという発想で作ったプラグインです。
もう自由に何でもかんでもやりたい場合は、filter-evalを使いましょう。

config

# ...

filters:
  - type: ruby_proc
    requires:
      - cgi
    columns:
      - name: data
        proc: |
          ->(data) do
            data["events"] = data["events"].map.with_index do |e, idx|
              e.tap { |e_| e_["idx"] = idx }
            end
            data
          end
      - name: id
        proc: |
          ->(id) do
            id * 2
          end
        type: string
      - name: comment
        proc: |
          ->(comment, record) do
            return [record["account"].to_s].to_json unless comment
            comment.upcase.split(" ").map { |s| CGI.escape(s) }
          end
        skip_nil: false
        type: json
    target: events

# ...

procという項目にrubyのprocのソースコードをまんま記述します。
第一引数に、そのカラムの値が入ります。
第二引数も受け取る様になってる場合は、レコード全体も受け取るようになっているので、他のカラムの値を使って加工できます。
procの返り値でカラムの内容が置き換えられるので、副作用を使って更新したい場合は、tapを使うなり返り値を指定するなりで調整します。

filter後のカラムの型はtypeで指定します。

skip_nilにfalseを指定すると、カラムの値がnilでもprocが動作します。
デフォルトではnilはスキップして処理しないようになっています。

requiresに配列で文字列を渡しておくと、事前にその名前のライブラリをrequireします。

これで大体の加工はできるはずだし、evalするよりは早く済む気がします。
要求されるパフォーマンス次第ですが、もし使い道があれば御活用ください。