Embulkで手軽にメンテナンス性高いETLバッチを書く

やりたいこと

普段はDDDxScalaでアプリケーションを書いていますが、
細かい粒度の大量データを画面表示用のView用テーブルに一定の粒度でまとめたいことがあります。

しかし、サマリバッチを書くとなると開発コスト&メンテナンス性等でツラミが出てくることが多くて辛くなりがち。

embulkなら
開発コスト: クエリ書くだけ
メンテナンス: クエリ直すだけ
なので、いい感じに書けないかやってみました。

環境構築

1. embulk.jarを落としてくる

$ curl --create-dirs -o ~/bin/embulk -L http://dl.embulk.org/embulk-latest.jar
$ chmod +x ~/bin/embulk
$ export PATH=$PATH:~/bin/

2. bundle用ディレクトリ生成

$ embulk mkbundle vendor/bundle

3. Gemfileに必要なパッケージ記述

$ cat vendor/bundle/Gemfile
source 'https://rubygems.org/'
gem 'embulk', '~> 0.8.0'
gem 'embulk-input-mysql',  '~> 0.7.1'
gem 'embulk-output-mysql', '~> 0.6.1'

4. bundle install

$ cd vendor/bundle
$ embulk.jar bundle install
$ cd ../../

Gemfileまで作れば次回以降はjar落としてbundle installするだけなのでRakefileでタスクに落としました。

desc "embulk init"
task :embulk_init do
  sh %Q(curl --create-dirs -o ~/bin/embulk -L http://dl.embulk.org/embulk-latest.jar \
        chmod +x ~/bin/embulk)
  Dir.chdir('./vendor/bundle') do
    sh %Q(embulk bundle install)
  end
end

liquid

ymlファイル内で制御構文、環境変数指定が使えます。

Example

in:
{% if env.EMBULK_ENV == 'production' %}{% include 'db/prod_slave' %}
{% elsif env.EMBULK_ENV == 'staging' %}{% include 'db/stg' %}
{% else %}{% include 'db/dev' %}
{% endif %}
{% include 'query/hoge'
  target_table: 'table_a',
  starts_at: env.TARGET_START,
  ends_at: env.TARGET_END %}

out:
{% if env.EMBULK_ENV == 'production' %}{% include 'db/prod_master' %}
{% elsif env.EMBULK_ENV == 'staging' %}{% include 'db/stg' %}
{% else %}{% include 'db/dev' %}
{% endif %}
  table: table_b
  mode: merge_direct

実行環境指定(dev,stg,prod)

if文が使えるのでこの部分で行っている。
環境変数EMBULK_ENVの内容でincludeするconfigを判定している。

{% if env.EMBULK_ENV == 'production' %}{% include 'db/prod_slave' %}
{% elsif env.EMBULK_ENV == 'staging' %}{% include 'db/stg' %}
{% else %}{% include 'db/dev' %}
{% endif %}

includeするファイルには命名規則が有ります。

• プレフィックス_をつける
• .yml.liquid拡張子をつける

db/_dev.yml.liquid

実行内容指定(クエリ等)

クエリに変数を埋め込めるので、実行対象期間を指定できるようにした。

{% include 'query/hoge' # 読み込むファイル指定
  target_table: 'table_a', # 以下includeするファイルに渡す引数
  starts_at: env.TARGET_START,
  ends_at: env.TARGET_END %}

渡した引数はincludeしたファイル内で以下のように呼び出せる

SELECT * FROM {{ target_table }}

inputの構造化されたデータに対して汎用的なバッチを書いたり、期間指定したりするのに
引数指定&制御構文は必須なのでembulkで出来たのは最高でした。
バッチ書くのに困ることはあまりなさそうです。

Rakefile

都度、環境変数を指定するのは面倒なのでRakeコマンドでまとめました。

desc "run all"
task :run_all, 'env', 'target_start', 'target_end'
task :run_all do |task, args|
  Dir.chdir('./') do
    sh %Q(rake run_hoge['#{args.env}','#{args.target_start}','#{args.target_end}'])
    sh %Q(rake run_bar['#{args.env}','#{args.target_start}','#{args.target_end}'])
  end
end

desc "preview hoge"
task :preview_hoge, 'env', 'target_start', 'target_end'
task :preview_hoge do |task, args|
  Dir.chdir('./') do
    sh %Q(export EMBULK_ENV=#{args.env} && \
          export TARGET_START=#{args.target_start} && \
          export TARGET_END=#{args.target_end} && \
          java -jar embulk.jar -b vendor/bundle preview ./cfg_hoge.yml.liquid -G)
  end
end

desc "run hoge"
task :run_hoge, 'env', 'target_start', 'target_end'
task :run_hoge do |task, args|
  Dir.chdir('./') do
    sh %Q(export EMBULK_ENV=#{args.env} && \
          export TARGET_START=#{args.target_start} && \
          export TARGET_END=#{args.target_end} && \
          java -jar embulk.jar -b vendor/bundle run ./cfg_hoge.yml.liquid)
  end
end

...

こんな感じで実行

rake run_all['staging','2016-07-01','2016-07-02']

罠

• 先頭に改行いれないとincludeのインデントがバグる
  これに盛大にハマりました。
  {% include 'db/dev' %}
  とするとdb/_dev.yml.liquidを呼び出せる。

当初以下のように書いていたが何故か呼べず。
インデントした位置に展開してくれるはずなのでyml的にも問題ないはずですが、syntaxエラーがでます。

{% if env.EMBULK_ENV == 'production' %}
  {% include 'db/prod_slave' %}
{% else if env.EMBULK_ENV == 'staging' %}
  {% include 'db/stg' %}
{% else %}
  {% include 'db/dev' %}
{% endif %}

includeで呼ぶと先頭一行のインデントがずれてしまうのが原因でした。
include対象ファイルの先頭に改行を入れることで回避可能。
※@hiroysatoさんに助けてもらいました。ありがとうございました。
http://qiita.com/hiroysato/items/861e3689eef430f5e723

db/_dev.yml.liquid

#　　 ※目印のためにコメントアウトして改行
type: mysql
host: 127.0.0.1
user: root
password: "hoge"
database: dev

インデントをcfgではなく、includeするファイル側でしてもOK

{% if env.EMBULK_ENV == 'production' %}{% include 'db/prod_slave' %}
{% elsif env.EMBULK_ENV == 'staging' %}{% include 'db/stg' %}
{% else %}{% include 'db/dev' %}
{% endif %}

db/_dev.yml.liquid

  type: mysql
  host: 127.0.0.1
  user: root
  password: "hoge"
  database: dev

• configファイルはembulkディレクトリ直下じゃないとバグる（パス指定方法の問題かも

embulk実行dir/configとして、実行するとコケた。
embulk実行dir直下にconfigファイルを置くことで回避。
includeのパス問題かも？※あまり追ってません。
相対パス指定{% include '../db/dev' %}してみたがコケたので諦めた

Jenkinsで定期実行

ソースコード管理にgithub指定してworkspaceディレクトリをDocker環境にマウントして実行

#!/bin/bash -xe

TARGET_START=${TARGET_START:-`date '+%Y-%m-%d' -d "1 day ago"`}
TARGET_END=${TARGET_END:-`date +%Y-%m-%d`}
ENV="production" # development|staging|production

IMAGE="" # 作成したイメージを入れてください

docker pull $IMAGE
docker run -t -v $WORKSPACE:/project $IMAGE sh -c "ls -l project && cd project/embulk && rake embulk_init && rake run_all['$ENV','$TARGET_START','$TARGET_END']"

Dockerfileはruby,rake,javaが実行できればOK