digdag-3tips

Transcript

1. Digdagを利用したデータ基盤と
 その運用Tips
 株式会社ZOZOテクノロジーズ
 SRE部 MA基盤 リーダー
 田島 克哉 Copyright © ZOZO Technologies, Inc.
2. © ZOZO Technologies, Inc. 株式会社ZOZOテクノロジーズ
 SRE部
 MA基盤 リーダー 田島 克哉
 2018新卒入社
 業務ではデータ基盤・配信基盤の開発運用を行っている
 Digdag/Embulk大好き系エンジニア
 好きな言語はClojure(得意ではない)
 
 2
3. © ZOZO Technologies, Inc. 目次
 • Digdag利用事例の紹介
 ◦ データ基盤
 • 運用Tips3選
 ◦ ワークフローの階層化
 ◦ エラーハンドリング
 ◦ ワーカーの自動スケールイン
 • まとめ
 3
4. © ZOZO Technologies, Inc. 目次
 • Digdag利用事例の紹介
 ◦ データ基盤
 • 運用Tips3選 ←メインコンテンツ
 ◦ ワークフローの階層化
 ◦ エラーハンドリング
 ◦ ワーカーの自動スケールイン
 • まとめ
 4
5. © ZOZO Technologies, Inc. https://zozo.jp/
 • 日本最大級のファッション通販サイト
 • 1,200以上のショップ、7,300以上のブランドの取り扱い（ともに2019年6 月末時点）
 • 常時73万点以上の商品アイテム数と毎日平均3,200点以上の新着商品 を掲載
 • 即日配送サービス
 • ギフトラッピングサービス
 • ツケ払い など
 5
6. © ZOZO Technologies, Inc. https://wear.jp/
 6 • 日本最大級のファッションコーディネートアプリ
 • 1,300万ダウンロード突破、コーディネート投稿総数は900万件以上（と もに2019年9月末時点）
 • 全世界（App Store / Google Playが利用可能な全ての国）でダウンロー ドが可能
 • 200万人以上のフォロワーを持つユーザー（WEARISTA）も誕生

7. © ZOZO Technologies, Inc. https://zozo.jp/multisize/
 
 • 身長と体重を選択するだけで理想のサイズの商品が見つかる新しい洋 服の買い方
 • ZOZOSUITで得た100万件以上の体型データを活用し、20～50サイズ のマルチサイズ（多サイズ）に展開
 • 2019年秋冬アイテムから、人気ブランドのマルチサイズアイテムを販売 開始
 【参加企業】
 　株式会社アーバンリサーチ、株式会社ストライプインターナショナル、
 　株式会社デイトナ・インターナショナル、株式会社パル、株式会社ビームス、
 　株式会社ベイクルーズ、MARK STYLER株式会社、リーバイ・ストラウス ジャパン株式会社　　など
 7
8. © ZOZO Technologies, Inc. Digdag利用事例紹介
 8
9. © ZOZO Technologies, Inc. ZOZOTOWNのデータ基盤
 • データ基盤の役割
 ◦ 複数のDBからBigQueryにデータを連携
 ◦ BigQuery上にデータマートを構築
 9
10. © ZOZO Technologies, Inc. データ基盤でのDigdagの役割
 10
11. © ZOZO Technologies, Inc. よくある質問
 • なんでDWHがBigQueryなのにAWS使ってるの？
 ◦ 歴史的背景が大きい
 ◦ もともとDWHにRedshiftを利用していた
 ◦ AWSでのDigdagの運用実績があった
 • データ連携頻度は？
 ◦ 1日に1回
 ◦ 現在リアルタイムデータ連携基盤を別途作成中
 • より詳しい基盤の構成等々が聞きたい
 ◦ コロナが明けたらオフィス遊びにきてくださいー
 ◦ リモートでもお話できます
 11
12. © ZOZO Technologies, Inc. 運用Tips3選
 12
13. © ZOZO Technologies, Inc. Tips1 ワークフローの階層化
 13
14. © ZOZO Technologies, Inc. データ基盤のワークフロー
 ワークフロー
 • データ連携
 
 • 差分更新
 
 • データマート更新
 
 14
15. © ZOZO Technologies, Inc. 各処理のワークフロー
 • データ連携
 ◦ 前処理
 ◦ データを連携するテーブルの一覧を取得
 ◦ テーブルごとにデータ連携
 • 差分更新
 ◦ 前処理
 ◦ 差分対象テーブル取得
 ◦ 差分更新処理
 • データマート更新
 ◦ 前処理
 ◦ 更新するデータマートの一覧を取得
 ◦ データマート更新
 
 15
16. © ZOZO Technologies, Inc. • データ連携
 ◦ 前処理
 ◦ データを連携するテーブルの一覧を取得
 ◦ テーブルごとにデータ連携
 • 差分更新
 ◦ 前処理
 ◦ 差分対象テーブル取得
 ◦ 差分更新処理
 • データマート更新
 ◦ 前処理
 ◦ 更新するデータマートの一覧を取得
 ◦ データマート更新
 
 多段階ワークフロー
 16 親ワークフロー 子ワークフロー 子ワークフロー 子ワークフロー
17. © ZOZO Technologies, Inc. ワークフローを階層化するメリット
 17 • 責任の分割ができる
 
 • ワークフローの再利用が可能
 
 • ワークフローごとに単独での実行が可能になる
 
 
 
 基本的にはプログラミングで言うところのサブルーチンと同じ
18. © ZOZO Technologies, Inc. Digdagで多段階ワークフローを実現する方法
 18 • 多段階ワークフローの実現方法4つ
 ◦ callオペレータの利用
 ◦ !includeの利用
 ◦ requireオペレータの利用
 ◦ APIを使った実装

19. © ZOZO Technologies, Inc. 実装例
 • 以下の子ワークフローを順番に実行するワークフローを考えていく
 
 19 +task_a: sh>: echo "workflow a" +sleep: sh>: sleep 5 +task_a: sh>: echo "workflow b" +sleep: sh>: sleep 5 +task_a: sh>: echo "workflow c" +sleep: sh>: sleep 5 a.dig b.dig c.dig
20. © ZOZO Technologies, Inc. callオペレータを使った多段ワークフロー
 20 +a: call>: a.dig +b: call>: b.dig +c: call>: c.dig abc.dig 結果
21. © ZOZO Technologies, Inc. !includeを使った多段ワークフロー
 21 +a: !include: a.dig +b: !include: b.dig +c: !include: c.dig abc.dig 結果
22. © ZOZO Technologies, Inc. !include/callオペレータの利点・欠点
 • 利点
 ◦ 1つのワークフローに処理がまとまっていてシンプル
 ◦ ワークフローの記述がシンプル
 • 欠点
 ◦ ログがめちゃでかくなる
 ◦ 子ワークフローの単位でretry allができない
 ▪ できるが後続の処理も手動で順次実行してやる必要がある
 • !includeとcallの違い
 ◦ スコープが違う
 ◦ 参考: https://qiita.com/pilot/items/6d03069388e5df5f4f40
 22
23. © ZOZO Technologies, Inc. !include/callオペレータの利点・欠点
 • 利点
 ◦ 1つのワークフローに処理がまとまっていてシンプル
 ◦ ワークフローの記述がシンプル
 • 欠点
 ◦ ログがめちゃでかくなる → WebUIが重くて開けない
 ◦ 子ワークフローの単位でretry allができない
 ▪ できるが後続の処理も手動で順次実行してやる必要がある
 • !includeとcallの違い
 ◦ スコープが違う
 ◦ 参考: https://qiita.com/pilot/items/6d03069388e5df5f4f40
 23
24. © ZOZO Technologies, Inc. requireを使った多段ワークフロー
 24 +a: require>: a +b: require>: b +c: require>: c abc.dig 結果
25. © ZOZO Technologies, Inc. requireオペレータの利点・欠点
 • 利点
 ◦ 親ワークフローは全体の実行の管理しかしないため見通しが良い
 ◦ ワークフローごとにWebUIが分割される
 • 欠点
 ◦ 子ワークフローが失敗して親ワークフローをリトライすると子ワークフローが最初から再 実行されてしまう(v0.9.41まで)
 25
26. © ZOZO Technologies, Inc. requireのリトライの問題再現
 • 子ワークフローで最後のジョブでエラーを発生させる
 26 +parent: require>: chiled parent.dig +echo_start: sh>: echo "start chiled" +sleep2: sh>: sleep 10 +echo_end: sh>: ech "end chiled" chiled.dig typo 結果 parent chiled
27. © ZOZO Technologies, Inc. リトライの問題についての再現
 • chiled.digを修正して親ワークフローをリトライ
 27 parent chiled 1回目で成功していたecho_startと sleep2も実行される 結果
28. © ZOZO Technologies, Inc. requireオペレータのリトライ問題の回避策
 • 実行(run)と待機(wait)の分割
 ◦ 親ワークフロー側で子ワークフローをrunするジョブとwaitするジョブを分ける
 • 親ワークフロー側の処理
 ◦ runでは子ワークフローを実行するだけ
 ◦ waitでは子ワークフローが成功するまでただただ待機
 ◦ 子ワークフローが成功したら後続の処理に続く
 • 子ワークフローが失敗した場合
 ◦ 親ワークフローはwaitをし続けるだけ
 ◦ 子ワークフローは「retry failed」または「retry all」にてリトライ
 
 
 
 
 28 ・ ・ ・
29. © ZOZO Technologies, Inc. 理想のリトライ方法
 29 abc.dig(親ワークフロー) RETRY FAILED b.dig(子ワークフロー)
30. © ZOZO Technologies, Inc. 理想のリトライ方法の実現方法
 
• digdagコマンドを使って子ワークフローをstart
 
 
 
 
 
 • APIを利用して子ワークフローの状態を監視して成功を待つ
 30 ・ ・ ・ run_b / run_cと続く
31. © ZOZO Technologies, Inc. APIを使った多段ワークフロー(実装)
 31 _export: project_name: sample digdag_endpoint_from_docker: 'http://host.docker.internal:65432' +run_a: +start: _retry: 3 sh>: tasks/digdag_start_if_session_not_exist.sh ${project_name} a "${session_local_time}" +wait: _retry: 3 _export: docker: image: 'ruby' require: 'tasks/wait_for_other_workflow' rb>: WaitForOtherWorkflow.wait workflow_name: a timeout: 60 digdag_endpoint: ${digdag_endpoint_from_docker} ・ ・ ・ run_b / run_cと続く
32. © ZOZO Technologies, Inc. 子ワークフロー起動スクリプト(bash)
 • 子ワークフローがすでに起動済みでないかを確認
 • 親ワークフローと同じsession_timeで子ワークフローを起動
 32 #!/bin/bash project_name=$1 workflow_name=$2 session_local_time=$3 digdag sessions ${project_name} ${workflow_name} | grep "session time: ${session_local_time}" > /dev/null if [ $? -ne 0 ]; then digdag start ${project_name} ${workflow_name} --session "${session_local_time}" fi
33. © ZOZO Technologies, Inc. 子ワークフローwaitスクリプト(ruby)
 33 ・ ・ ・ run_b / run_cと続く require 'json' require 'faraday' class WaitForOtherWorkflow def wait(workflow_name:, timeout:, interval: 1, digdag_endpoint: 'http://localhost') session_time = Digdag.env.params['session_time'] limit = Time.now + timeout loop do sessions = get_sessions(digdag_endpoint) target_session = sessions.find {|session| session['workflow']['name'] == workflow_name && session['sessionTime'] == session_time } case session_state(target_session) when :success break when :not_started, :pending, :failure raise 'TIMEOUT' if Time.now > limit sleep(interval) else raise 'invalid session state' end end end private def get_sessions(digdag_endpoint) JSON.parse(Faraday.get("#{digdag_endpoint}/api/sessions").body)['sessions'] end def session_state(session) return :not_started if session.nil? last_attempt = session['lastAttempt'] if last_attempt['done'] if last_attempt['success'] :success else :failure end else :pending end end end • 子が成功した場合
 ◦ 親ワークフローのジョブを成功に
 • 子が失敗・実行中の場合
 ◦ 親ワークフロー側はTimeoutまでwait

34. © ZOZO Technologies, Inc. APIを使った多段ワークフロー(成功結果)
 34
35. © ZOZO Technologies, Inc. リトライ時(再掲)
 35 abc.dig(親ワークフロー) RETRY FAILED b.dig(子ワークフロー)
36. © ZOZO Technologies, Inc. APIを使うメリット・デメリット
 • 利点
 ◦ 親ワークフロー側の全体の見通しがよくなる
 ◦ ログが子ワークフローごとに分割される
 ◦ ワークフローごとに柔軟なリトライが可能
 • 欠点
 ◦ 実装を作り込む必要がある
 ◦ APIの仕様が変わったら動かなくなる
 36
37. © ZOZO Technologies, Inc. 各手法の比較
 37 APIを使った多段ワークフローを利用 call / !include require API WebUIの使い勝手 △(データ量による) ◎ ◎ リトライ △ ☓ ◎ メンテナンス性 ◎ ◎ △
38. © ZOZO Technologies, Inc. Digdag 0.9.42での改善
 • #1379
 ◦ WebUIログ重い問題の解決改善
 • #1422
 ◦ requireのリトライ問題の改善
 38 https://docs.digdag.io/releases/release-0.9.42.html
39. © ZOZO Technologies, Inc. WebUI重い問題の改善[#1379]
 • REFRESH_LOGSボタンの追加
 ◦ WebUIのログがボタンを押すまでrefreshされなくなった
 ◦ これでWebUIのログ展開が重い問題が解消されてた！
 
 • TimeLineの展開が重い
 ◦ ログだけの問題かと思ったらTimeLineの展開も重かった
 
 
 39
40. © ZOZO Technologies, Inc. requireのリトライ問題の改善[#1422]
 • require_onオプションの追加
 ◦ requireオペレータ使用時リトライの子ワークフローの挙動を選べるように
 
 40 https://docs.digdag.io/operators/require.html#options
41. © ZOZO Technologies, Inc. Tips2 エラーハンドリング
 41
42. © ZOZO Technologies, Inc. エラーハンドリング
 よくやるエラーハンドリング
 • _errorを使ってエラーをcatch
 • SlackプラグインでSlackに通知
 ◦ https://github.com/szyn/digdag-slack
 
 42 _error: _retry: 3 slack>: failed-to-workflow.yml
43. © ZOZO Technologies, Inc. _errorの問題点
 • Digdagのワーカーがjobを目一杯つかんでいると_errorが実行されない
 ◦ max-taskthreads分のjobをワーカーが処理していると_errorが実行されない
 
 • _export内でエラーが発生すると実行されない(#1203)
 ◦ _export内でエラーが発生すると無言でワークフローが死ぬ
 
 • すべてのワークフローに対して_errorを書いて回らないといけない
 ◦ _errorを書き忘れると無言でワークフローが死ぬ
 
 • _error内でエラーが発生すると通知されない
 ◦ _error内で問題が発生した場合通知に失敗する
 43
44. © ZOZO Technologies, Inc. notification機能を使う
 
 • エラーが発生したときに以下を実行
 ◦ shell command
 ◦ httpリクエスト
 ◦ mail送信
 • 参考
 ◦ #669のissueにて古橋さんも推奨
 44 notification.type = shell notification.shell.command = echo "error!!!!" notification.type = http notification.http.method = GET notification.http.url = https://example.com notification.http.headers.... = .... notification.type = mail notification.mail.from = ... notification.mail.to = ... notification.mail.cc = ... notification.mail.bcc = ... notification.mail.subject = ... notification.mail.body_template_file = ... notification.mail.html = ...
45. © ZOZO Technologies, Inc. notification機能の問題点
 • 1つのdigdagに1つの設定
 ◦ ワークフローごとに柔軟なメッセージの作成が 難しい(shellでif文)
 • 以下の情報が取得できない
 ◦ Error Message
 ◦ StackTrace
 
 45 { "timestamp": "2020-06-22T15:40:04Z", "message": "Workflow session attempt failed", "site_id": 0, "project_id": 1, "project_name": "sample", "workflow_name": "workflow", "revision": "22ee5636-9b60-41e0-9465-fda5ab41213e", "attempt_id": 1, "session_id": 1, "task_name": "+workflow^failure-alert", "timezone": "UTC", "session_uuid": "60372a5f-d5a1-4cd3-b2e9-d98edb47bf73", "session_time": "2020-06-22T15:40:03+00:00" } notification内で使える情報
46. © ZOZO Technologies, Inc. _errorとnotification機能の使い分け
 • notification
 ◦ Slackへはメンション付きで通知する
 ◦ エラーの発生元がわかるくらいの範囲の情報を通知
 ◦ PagerDutyに対しても合わせて通知を行う
 • _error
 ◦ Slackへはメンションなしで通知する
 ◦ なるべく詳しい情報を通知
 ◦ ワークフロー固有のエラーメッセージもここで定義
 46
47. © ZOZO Technologies, Inc. notification.shellの例
 47 JSON=$(cat -) WORKFLOW_NAME=$(echo $JSON | jq ".workflow_name" -r) MESSAGE=$(echo $JSON | jq ".message" -r) ATTEMPT_ID=$(echo $JSON | jq ".attempt_id" -r) SESSION_ID=$(echo $JSON | jq ".session_id" -r) SLACK_PAYLOAD=$(cat << EOS payload={"channel": "#digdag_notice", "username": "webhookbot", "attachments": [{"color": "danger", "title": "<!channel>\n[ERROR] $WORKFLOW_NAME", "text": "$MESSAGE", "fields": [{"title": "Session", "value": "https://example.com/sessions/$SESSION_ID", "short": false}, {"title": "Attempt", "value": "https://example.com/attempts/$ATTEMPT_ID", "short": false}]}]} EOS ) PAGERDUTY_PAYLOAD=$(cat << EOS { "routing_key": "xxxxxxxxxxxxxxxxxxxxxxxxxxxx", "event_action": "trigger", "payload": { "summary": "[ERROR] Digdag $WORKFLOW_NAME", "source": "Digdag", "severity": "error" } } EOS ) set +e echo "[Slack]\n${SLACK_PAYLOAD}" curl -X POST --data-urlencode "${SLACK_PAYLOAD}" https://hooks.slack.com/services/xxxxxxxxxxxxxxxxxxxxx echo "[PagerDuty]\n${PAGERDUTY_PAYLOAD}" curl -X POST -d "${PAGERDUTY_PAYLOAD}" https://events.pagerduty.com/v2/enqueue set -e
48. © ZOZO Technologies, Inc. Tips3 自動スケールイン
 48
49. © ZOZO Technologies, Inc. Digdagのアーキテクチャ
 • Jobキュー型のアーキテクチャ
 49
50. © ZOZO Technologies, Inc. Digdagワーカーのスケーリング
 50
51. © ZOZO Technologies, Inc. Digdagワーカーのスケーリングの問題
 • スケールアウト
 ◦ サーバーが増えてDigdagが起動するとjobを実行できるようになる
 ◦ 何も問題なし
 
 • スケールイン
 ◦ スケールインするサーバーのDigdagがjobを掴んでいたらjobが失敗する
 ◦ jobが失敗した場合リトライすればよいが処理が長い場合はリトライしたくない
 ◦ ワーカーがjobを掴んでいないタイミングでスケールインする必要がある
 51 どうやってワーカーがjobを掴んでいないか判断する?
52. © ZOZO Technologies, Inc. queued_task_locksテーブル
 • queued_task_locksテーブル
 ◦ 実行中のjobを管理するテーブル
 ◦ lock_agent_idに実行中のホスト名が含まれている
 52 digdag=> select * from queued_task_locks; id | site_id | queue_id | priority | retry_count | lock_expire_time | lock_agent_id ------+---------+-------------+---------+----------------+-----------------------+------------------------------------- 1176 | 0 | | 0 | 0 | 1591769498 | 19349@digdag-worker-1 ここに含まれていないhostを停止してやれば良さそう！
53. © ZOZO Technologies, Inc. 使っていないワーカーself shutdown作戦
 • HARAKIRIスクリプト
 ◦ lock_agent_idの一覧を取得して来て自分自身のhost名が含まれなければshutdownを する
 • cronで定期実行をする
 ◦ Digdagで実行するとHARAKIRIスクリプトのジョブが動くのでself shutdownできない
 53
54. © ZOZO Technologies, Inc. HARAKIRIスクリプト(例: EC2)
 54 export PGPASSWORD='ポスグレのパスワード' POSTGRESQL_HOST_NAME='ポスグレのホスト名' POSTGRESQL_USER_NAME='ポスグレのユーザー名' RETRY_COUNT='リトライ回数' RETRY_INTERVAL='リトライ間隔' MYHOST=`hostname` for i in `seq 1 $RETRY_COUNT` do HOST_LIST=`/usr/bin/psql -t -h POSTGRESQL_HOST_NAME -U POSTGRESQL_USER_NAME -c 'select lock_agent_id from queued_task_locks;'` for HOST in $HOST_LIST do if [ $(echo $HOST | grep -e $MYHOST) ]; then exit 0 fi done sleep $RETRY_INTERVAL done MYINSTANCEID=`curl 169.254.169.254/latest/meta-data/instance-id/` REGION='AWSのリージョン' AUTOSCALING_GROUP='オートスケーリンググループの ARN' /usr/local/bin/aws autoscaling detach-instances --instance-ids $MYINSTANCEID --auto-scaling-group-name $AUTOSCALING_GROUP --should-decrement-desired-capacity --region $REGION /usr/local/bin/aws ec2 terminate-instances --instance-ids $MYINSTANCEID --region $REGION ←たまたまjobを掴んでいないケースもあるため何回かリトライ ←自分のホスト名が含まれていたら終了 ↓自分のホスト名が含まれていなかった場合自分自身を Shutdown
55. © ZOZO Technologies, Inc. HARAKIRIスクリプトを使った問題点
 • cronを管理する必要がある
 ◦ せっかくDigdagを使っているのに別途cronの管理が必要になる
 • 必要以上にスケールインしてしまう
 ◦ 処理時間が長い単一のタスクがあるとスケールインされてしまう
 
 55 タスク2実行中にスケールインされてしまう 例:
56. © ZOZO Technologies, Inc. HARAKIRIスクリプトを使うと良い場面
 • 複数のワークフローが同時に動く
 ◦ 単一のワークフローしか動かない場合はワークフローの実行が終わったタイミングでス ケールインすればOK
 • ワークフローごとにワーカーが必要な数にむらがある
 ◦ ワークフローごとにスケール数が固定であれば0/1条件になるためワークフローが動 作しているかどうかだけで判断が可能
 56
57. © ZOZO Technologies, Inc. HARAKIRIスクリプトを使うと良い場面
 • 複数のワークフローが同時に動く
 ◦ 単一のワークフローしか動かない場合はワークフローの実行が終わったタイミングでス ケールインすればOK
 • ワークフローごとにワーカーが必要な数にむらがある
 ◦ ワークフローごとにスケール数が固定であれば0/1条件になるためワークフローが動 作しているかどうかだけで判断が可能
 57 最終手段として使ってください
58. © ZOZO Technologies, Inc. • 以下のTipsを紹介しました
 ◦ ワークフローの階層化
 ◦ エラーハンドリング
 ◦ ワーカーの自動スケールイン
 • Digdagを運用してみての感想
 ◦ 機能をフル活用するにはコードレベルまでの把握が必要
 ◦ もっとチームメンバーが増えればもっとコントリビュートできる
 • Digdagはいいぞ
 ◦ Digdagめちゃめちゃ便利!!
 58 まとめ

59. © ZOZO Technologies, Inc. 最後に
 Digdagメンテナの皆様いつもありがとうございます
 
 59