digdagをDockerizeしてECS上で運用することにしました

データ分析や可視化に伴う複雑なジョブフローの改善にはdigdagが便利です。
少しずつ採用事例も増えているようです。

qiita.com

今回は、そんな便利なdigdagをECS上に構築しました。

*1

事前知識

digdagに関する基本的な知識は、以前のエントリを参考にしてください。

yukiyan.hatenablog.jp

コード

サンプル用にコードを公開しました。

github.com

digdagをDockerizeし、設定ファイル(digファイル)も一緒に固めてECRにpushしています。
つまり、digdagの最新の設定ファイルは常にECRにある状態です。
digdagの設定ファイルを変更したブランチがmasterにマージされると、shippableがdocker build・digdag check・docker push・ECSに関する処理をおこない、古いdigdagコンテナが破棄され、新しいdigdagコンテナが立つというライフサイクルです。
開発者の作業はGitHubだけで完結します。
現状、登録するジョブがまだ少ないので、無停止運用を頑張らずdigdagコンテナを毎回新たに立てるという運用にしています。
ちなみに、digdagには指定したdocker imageでタスクを実行したり、digdag pushという設定ファイルを登録する機構もあるので、それらを活用すれば無停止運用も可能です。

選定理由

ECSを使う理由

今年1月にJUBILEE WORKSという会社に転職をしたのですが、そこでは基本的に全ての環境がECS上に構築されています。
したがって、なにかを導入するときはDockerizeしてECS上に立ててしまうのが一番作業コストが低いです。
「ECS上に立ててdigdagをスケーラビリティに！ホットデプロイ！無停止運用！」というより「コンテナをAWS上で動かすならECSが一番楽だよね」くらいの気軽な感覚です。

digdagを使う理由

今回、MySQLのデータをEmbulkでBigQueryに入れてRe:dashで可視化&分析をおこなううえで、 MySQLからBigQueryへの一連のワークフローを効率よく組むのにdigdagが最適でした。
digdagを使うと、各処理を並列で実行できるので便利です。
また、変数やfor_eachを使うと似たようなEmbulkのymlを書かなくて済むのも便利です。

shippableを使う理由

主にDockerコンテナのキャッシュ目的です。
それ以外の機能はだいたいどこのCIサービスでも備えているものが多いです。

構築時のポイント

digdagのdockerize

以下のようになりました。

FROM java:8

MAINTAINER yukiyan <[email protected]>

ENV DIGDAG_VERSION=0.9.3

RUN apt-get update && apt-get upgrade -y && apt-get install -y --no-install-recommends \
      jruby && \
    curl -o /usr/local/bin/digdag --create-dirs -L "https://dl.digdag.io/digdag-${DIGDAG_VERSION}" && \
    chmod +x /usr/local/bin/digdag && \
    curl -o /usr/local/bin/embulk --create-dirs -L "http://dl.embulk.org/embulk-latest.jar" && \
    chmod +x /usr/local/bin/embulk && \
    apt-get clean && rm -rf /var/cache/apt/archives/* /var/lib/apt/lists/* && \
    adduser --shell /sbin/nologin --disabled-password --gecos "" digdag

USER digdag

WORKDIR /home/digdag

RUN embulk gem install \
      embulk-input-mysql \
      embulk-output-bigquery

COPY tasks tasks
COPY main-digdag.dig .

EXPOSE 65432

CMD ["java", "-jar", "/usr/local/bin/digdag", "scheduler", "-m"]

java:8-alpineを使いたかったのですが、digdag内でのEmbulkの処理の部分でうまくいかなかったので断念しました(そんなに問題ではない)。

あと、CMD ["digdag", "scheduler", "-m"]ではなくjava -jarで実行しないとエラーになります。 このissueが参考になります。

github.com

jrubyはembulk-input-mysqlのインストール時に必要なので入れてます。

また、digdagでRubyやPythonのオペレータを使う場合はそれも入れる必要があります。
今はまだEmbulkしか使ってないので入れてません。

digdagの設定ファイルの作成

「MySQLからBigQueryへEmbulkでロードする」というワークフローを組むと、以下のような設定ファイルになります。
単純な構造なので、なんとなく雰囲気は掴めると思います。

# main-digdag.dig
timezone: UTC

schedule:
  daily>: 00:00:00

+main:
  _export:
    host: 'mysql.hogehoge.ap-northeast-1.rds.amazonaws.com'
    user: 'hoge'
    password: 'fuga'
    project_id: 'bigquery_sample_project'
    dataset: 'bigquery_sample_dataset'

  +all_load:
    _parallel: true
    +load_db1:
      !include : 'tasks/db/db1.dig'
    +load_db2:
      !include : 'tasks/db/db2.dig'

メインのdigファイルです。
_export:は変数を定義しています。${host}のようにして参照できます。
!include:を使うとdigファイルを分割できます。
_parallel: trueを指定することで子タスクである+load_db1と+load_db2が並列で実行されます。孫タスクには影響しません。

# tasks/db/db1.dig
+mysql_bigquery:
  _export:
      database: 'db1'

  +load:
    for_each>:
      table: [
        hoge_table_1,
        hoge_table_2,
        hoge_table_3,
        hoge_table_4,
      ]
    _do:
      embulk>: tasks/db/embulk/mysql_bigquery.yml

# tasks/db/db2.dig
+mysql_bigquery:
  _export:
      database: 'db2'

  +load:
    for_each>:
      table: [
        sample_table_1,
        sample_table_2,
        sample_table_3,
        sample_table_4,
      ]
    _do:
      embulk>: tasks/db/embulk/mysql_bigquery.yml

子タスクのdigファイルです。
for_eachのおかげで以下のような冗長な記述をスマートに書けます。
for_each内も_parallel: trueで並列処理可能ですが、あまり並列度を増やしすぎるとlock系の警告が出るので、ここはあえて並列にしませんでした。

# 冗長な記述の例
+mysql_bigquery:
  _export:
      database: 'db1'

  +load:
    _export:
        table: 'hoge_table_1'
    embulk>: tasks/db/embulk/mysql_bigquery.yml
    _export:
        table: 'hoge_table_2'
    embulk>: tasks/db/embulk/mysql_bigquery.yml
    _export:
        table: 'hoge_table_3'
    embulk>: tasks/db/embulk/mysql_bigquery.yml
    _export:
        table: 'hoge_table_4'
    embulk>: tasks/db/embulk/mysql_bigquery.yml

# tasks/db/embulk/mysql_bigquery.yml
in:
  type: mysql
  host: ${host}
  user: ${user}
  password: '${password}'
  database: ${database}
  table: ${table}
out:
  type: bigquery
  mode: replace
  auth_method: json_key
  json_keyfile:
    content: |
      {
          "private_key_id": "123456789",
          "private_key": "-----BEGIN PRIVATE KEY-----\nABCDEF",
          "client_email": "..."
       }
  project: ${project_id}
  dataset: ${dataset}
  auto_create_table: true
  table: ${table}_${session_date_compact}
  allow_quoted_newlines: true

Embulkの設定ファイルです。
これまでのdigファイルの_exportで定義した変数を活用することで、似たようなEmbulkの設定ファイルを作る必要が無くなります。
あと、password:の箇所のみですが、${password}ではなく'${password}'にしないとエラーになります。これはたぶんembulk-input-mysqlのバグかも。

digdagのログ

digdagのログはデフォルトでは標準出力に吐かれるので、log driverを使ってcloudwatch logsに送っています。

digdag:
  image: 123456.dkr.ecr.ap-northeast-1.amazonaws.com/digdag:${BUILD_NUMBER}
  ports:
    - 65432:65432
  memory: 2000
  essential: true
  log_driver: awslogs
  log_opt:
    awslogs-group: /ecs/digdag
    awslogs-region: ap-northeast-1

shippableの設定ファイルの作成

イメージのbuild・ECRへのpush・ECSのTask definitionの登録やupdate-serviceはshippableに任せています。
shippableについては、r7kamuraさんの記事を参考にしてください。

r7kamura.hatenablog.com

ecs-formationについては社内で利用実績があるので使っています。
使い方は作者のstormcatさんの記事を参考にしてください。

blog.stormcat.io

language: go
go:
  - 1.5
env:
  global:
    - secure: hogehoge
build:
  ci:
    - export DOCKER_TAG=${ECR_REPOSITORY_URL}:${BUILD_NUMBER}
    - docker build --tag ${DOCKER_TAG} .
    - docker run ${DOCKER_TAG} java -jar /usr/local/bin/digdag check
  post_ci:
    - >
      go get github.com/stormcat24/ecs-formation;
      echo -e "project_dir: $SHIPPABLE_BUILD_DIR/ecs-formation\naws_region: ap-northeast-1" > $HOME/.ecs-formation.yml
    - ecs-formation task plan -p BUILD_NUMBER=${BUILD_NUMBER} -t digdag
    - >
      if [[ "${BRANCH}" == "master" ]]; then
        sudo docker push ${DOCKER_TAG}
        ecs-formation task -p BUILD_NUMBER=${BUILD_NUMBER} apply -t digdag
        aws ecs update-service --cluster digdag --service digdag --task-definition digdag --deployment-configuration maximumPercent=100,minimumHealthyPercent=0
      fi
integrations:
  hub:
    - integrationName: ecr
      type: ecr
      region: ap-northeast-1
      branches:
        only:
          - master
  notifications:
    - integrationName: email
      type: email
      on_success: never
      on_failure: never
      on_pull_request: never
    - integrationName: slack
      type: slack
      on_failure: always
      on_success: always
      recipients:
        - "#sandbox"
      branches:
        only:
          - master

secure: hogehogeの中には、AWSのトークン等の秘匿値が暗号化されてます。

BUILD_NUMBERやSHIPPABLE_BUILD_DIRはshippableの定数で、他にもいくつかあります。

所感

今回はdigdagをschedulerとして常時起動するようにしていますが、今後AWS Batchで代替できそうな気がします。
なので、AWS Batchでdigdag runを定時実行させるのが結構良さそうな気がします。インスタンスの常時起動不要になるし、スポットインスタンスも活用しやすいし。日本リージョン来たら検証したいです。
AWS Batchの裏側はECSなので、ECSで動かせるということはAWS Batchでも動きます(おそらく...)。
あと、shippableはdockerコンテナをキャッシュしてくれるおかげですごく速くて快適です。

*1:ネット上ではyukiyan, 会社ではmikeという源氏名で活動しています。出勤時間が昼過ぎなので22時でもslackにいます。