たそらぼ

日頃思ったこととかメモとか。

digdagでpythonコードを呼び出す

収集基盤

久しぶりにdigdagを使うことになったら、完全に全てを忘れていたので、pythonとの連携方法を再確認した。

以下のdigdagドキュメントのLanguage API - Pythonの挙動を抜粋して確認した。
Language API - Python — Digdag 0.10.0 documentation

f:id:tasotasoso:20190731083120p:plain

ディレクトリ構成

windows環境でやったので、以下のような感じ。tree /Fの結果の抜粋。

C:.
│  workflow.dig
└─tasks
    │  __init__.py

digdagのワークフロー定義ファイル

pythonは>pyの行のフォーマットで呼び出す。
.でパッケージ構造を指定する。

  • workflow.dig
+step1:
  py>: tasks.MyWorkflow.step1

+step2:
  py>: tasks.MyWorkflow.step2

pythonのスニペット

ドキュメントによると、__init__.pyに定義を書いておくスタイルらしい。

  • __init__.py
import digdag

class MyWorkflow(object):
  def step1(self):
    print("step1")
    digdag.env.store({'my_value': 1})

  def step2(self):
    print("step2: %s" % digdag.env.params["my_value"])

1発動かして挙動を確認するには、workflow.digのあるディレクトリでdigdag runコマンドを使う。

digdag run ./workflow.dig

このスニペットを動かすだけで、pythonスニペットの呼び出しと、digdagとの変数のやり取りが確認できるので便利。


あたり前ではあるが、別のパッケージに処理内容を定義しておいて、__init__.pyからimportするようにしておいても動いた。

  • __init__.py
from .task_contents import *
  • task_contents.py
import digdag

class MyWorkflow(object):
  def step1(self):
    print("step1")
    digdag.env.store({'my_value': 1})

  def step2(self):
    print("step2: %s" % digdag.env.params["my_value"])

感想

前回使おうと思ったのが2年前だが、それから結構本格運用している企業が増えたようで、ブログなどの公開情報で品質が高いものが新しく公開されていた。
特に、Digdagからdockerコンテナを呼んだり、Digdagをコンテナで動かしたりなどのノウハウが貯まってきているのがすごい。しっかり読んで、参考にさせていただきたい。