Digdagスケジューラの概要と動作確認

図などは公式サイトから引用
https://docs.digdag.io/architecture.html

Digdagとは

Digdagは複雑なパイプラインのタスクを実行、スケジュール、モニタリングするためのツール。依存関係などをハンドルすることができる。
DigdagはDAG(Directed Acyclic Graphs)を使用することにより、cronの置き換えや、ITオペレーションの自動化、データ分析のバッチジョブ、機械学習のパイプラインなどにマッチしている。
https://www.digdag.io/

Digdagのアーキテクチャ

ワークフロー自動化

ワークフローによりいくつかのマニュアルオペレーションを自動化することができる。ワークフローとしていくつかのタスクを定義し、Digdagはこれを実行する。タスクはoperatorプラグインを使用することで定義され、workflowエンジンでコントロールされる。、
Digdagはワークフロー自動化のためのフレームワークを提供することにより、開発者は自動化に集中することができる。もしタスクが失敗した場合、Digdagはアラートを送信する。またタスクが期待した時間内に終了しなかった場合、Digdagはnotificationを送信する。タスクはローカルマシン、分散サーバ、Dockerコンテナ上で実行することができる。

Groupでタスクを整理する

複雑なワークフローを自動化するとき、定義が複雑になりがちである。Diddagを使用することにより、タスクをグループに整理することができる。定義を確認するとき、概要を確認し、それから詳細化していく。これは開発においてデバッグやレビューを容易にする。これは運用環境では、管理者が何が起こって、どのように問題を修正するかについて理解するのに役立つ。

タスクが起動し、依存関係がないなら、全てのタスクが正常に完了する。
親グループが実行されるとき、それは子グループも実行する。そして親子全てが終了したら完了である。しかし、親グループが成功し、子グループが失敗した場合は、全体としては失敗になる。

パラメータ

グルーピングしたタスクはタスク間にパラメータを渡すことにも使用される。親タスクは子タスクの変数をエクスポートできる。（UNIXシェルのexportコマンドのように環境変数を設定）
親は子タスクを生成することができ、以前のタスクの結果に依存して異なるタスクを実行することができる。詳細は"defining variables"セクションを参照。

Workflow as Code

Digdagワークフローはコードで定義でき、これによりソフトウェア開発におけるベストプラクティス(version管理やレビュー、プルリクなど）をもたらしてくれる。

インストール

以下のコマンドを実行し、Digdagをインストールする。

$ curl -o /usr/local/bin/digdag --create-dirs -L "https://dl.digdag.io/digdag-latest"
$ chmod +x /usr/local/bin/digdag
$ digdag --help
2016-06-27 07:50:19 +0000: Digdag v0.8.2
Usage: digdag <command> [options...]
  Local-mode commands:
    new <path>                       create a new workflow project
    r[un] <workflow.dig>             run a workflow
    c[heck]                          show workflow definitions
・・・

実行

digdag initを実行し、サンプルのワークフローを生成する。

$ digdag init mydag
2016-06-27 07:51:52 +0000: Digdag v0.8.2
  Creating mydag/.gitignore
  Creating mydag/tasks/shell_sample.sh
  Creating mydag/tasks/repeat_hello.sh
  Creating mydag/tasks/__init__.py
  Creating mydag/mydag.dig
Done. Type `cd mydag` and then `digdag run mydag.dig` to run the workflow. Enjoy!

サンプルのmydag.digが作成される。3つのstepから構成されており、step1が終了するとstep2が実行、step2が全て終了するとstep3が実行される。step2は並列実行するように設定。

mydag.dig

timezone: UTC
_export:
  hello: "Hello, world!"
+step1:
  sh>: tasks/shell_sample.sh
+step2:
  _parallel: true
  +worker1:
    sh>: tasks/repeat_hello.sh
  +worker2:
    sh>: tasks/repeat_hello.sh
+step3:
  # defined at tasks/__init__.py
  py>: tasks.MyWorkflow.step3

$ cd mydag/
$ digdag run mydag.dig 
2016-06-27 07:52:32 +0000: Digdag v0.8.2
2016-06-27 07:52:34 +0000 [WARN] (main): Using a new session time 2016-06-27T00:00:00+00:00.
2016-06-27 07:52:34 +0000 [INFO] (main): Using session .digdag/status/20160627T000000+0000.
2016-06-27 07:52:34 +0000 [INFO] (main): Starting a new session project id=1 workflow name=mydag session_tim
e=2016-06-27T00:00:00+00:00
2016-06-27 07:52:35 +0000 [INFO] (0017@+mydag+step1): sh>: tasks/shell_sample.sh
Step1 of session 2016-06-27T00:00:00+00:00
2016-06-27 07:52:35 +0000 [INFO] (0017@+mydag+step2+worker1): sh>: tasks/repeat_hello.sh
Hello, world! from process 23680
2016-06-27 07:52:36 +0000 [INFO] (0019@+mydag+step2+worker2): sh>: tasks/repeat_hello.sh
Hello, world! from process 23682
Hello, world! from process 23680
Hello, world! from process 23682
Hello, world! from process 23680
Hello, world! from process 23682
Hello, world! from process 23680
Hello, world! from process 23682
2016-06-27 07:52:40 +0000 [INFO] (0019@+mydag+step3): py>: tasks.MyWorkflow.step3
Step3 of session 2016-06-27T00:00:00+00:00
Success. Task state is saved at .digdag/status/20160627T000000+0000 directory.
  * Use --session <daily | hourly | "yyyy-MM-dd[ HH:mm:ss]"> to not reuse the last session time.
  * Use --rerun, --start +NAME, or --goal +NAME argument to rerun skipped tasks.