目次 ワーカーノードの作成 DigdagとEmbulkのDockerビルド KubernetesにDigdag/Embulkをデプロイ Redashの導入 まとめ Kubernetes上に分析環境を構築する機会があったのでどのように構築したかを紹介します。同じような構成でKubernetes上で構築するのは3回目になったので構築方法も洗練されてきました。構成は以下のようになっています。 MySQL(RDS): サービスのデータベース。ここのテーブルからBigQueryにEmbulkでデータをエクスポートします。 PostgreSQL(RDS): Digdagのデータベース。今回新たにつくりました。 Digdag: データベースのエクスポートなどを実行するタスクスケジューラ。失敗したときにリトライもできます。 Embulk: プラグインを使ってデータベースをMySQLからBigQueryにエ

始めに 本記事ではTreasure Workflow(以後Workflow)における用語について説明します。 なんとなく使っていたけどよくわかっていない人や、使っていて何がなんだかわからなくなった人は読んでいただけますと幸いです。 Workflow と Digdagって何が違うの？ 本題に入る前に、WorkflowとDigdagの違いについて簡単に説明します。 DigdagはOSSのワークフローエンジンです。動作環境は自前で用意する必要があり、当然自分たちで運用しなければいけません。代わりに設定は自由にできますし、任意のタイミングでメンテナンス・バージョンアップなども可能です。 Workflowは簡単に言ってしまえばDigdagをTreasure Data(以下TD)がホストしているサービスです。作成したworkflowはTreasure Dataが運用しているサーバー上で動作するため、O

この記事はZOZOテクノロジーズ #5 Advent Calendar 2019 7日目の記事になります。 また、今年は全部で5つのAdvent Calendarが公開されています。 ZOZOテクノロジーズ #1 Advent Calendar 2019 ZOZOテクノロジーズ #2 Advent Calendar 2019 ZOZOテクノロジーズ #3 Advent Calendar 2019 ZOZOテクノロジーズ #4 Advent Calendar 2019 概要 Digdagのstoreパラメータに大きな値を入れたときに後続のジョブがすべて遅くなってしまうと言うことがあったので紹介します。 考えてみれば大きな値を毎回DBから取り出してDigdag側でパース等しているので重くなるのは当たり前なのですが。 そこで今回は実際にどんな感じで遅くなるのかを紹介します。また、その回方法を紹介

この記事はZOZOテクノロジーズ #5 Advent Calendar 2019 8日目の記事になります。 また、今年は全部で5つのAdvent Calendarが公開されています。 ZOZOテクノロジーズ #1 Advent Calendar 2019 ZOZOテクノロジーズ #2 Advent Calendar 2019 ZOZOテクノロジーズ #3 Advent Calendar 2019 ZOZOテクノロジーズ #4 Advent Calendar 2019 概要 DigdagのDBプラグインとしてPostgreSQLやMySQLが存在します。しかし使いたいDB(今回はPureData)に対応しているプラグインがありませんでした。 そこでJDBCドライバーさえあれば使えるdigdag-myjdbcというプラグインを作ろうとしたので紹介します。 なぜ名前をmyjdbcにしたかと言うと

だそうです。 簡単に言うと、INPUT元からデータを取り込み、OUTPUT先へデータを転送するデータローダーです。 Fluentd をご存知の人はイメージしやすいと思いますが、 Fluentd と同じく 取り込み元や転送先に合わせた plugin が各種用意されていて、そのプラグインを使用することでいろんなデータの保存先から別のデータ保存先へデータを転送することができます。 Fluentd がストリーム転送なのに対して、Embulk はバッチ形式でデータを転送します。 Embulk を使ったメリット 今回、なぜ Embulk を導入したのかというと、あらゆるデータ形式への柔軟な対応と今後のメンテナンス性の担保を可能にするために Embulk を導入しました。 データ形式への柔軟な対応 AI を運用する上で、お客さんからもらうデータは命のようなものですが、そのデータの形式や連携先のサービスな

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 f

Digdagワークフローのエラーを検知するためにトップレベルの _error を定義しslackに通知しています。 しかし _export 時にエラーになると _error のタスクが実行されないということがあったのでそれを紹介します。 こちらの記事のサンプルで利用しているDigdagのバージョンはすべて 0.9.39 です。 _errorについて _error を定義することで、Digdagのタスクが失敗した時に _error に定義されているタスクが実行されます。 以下が例で、task1の中で　sh>: ech "hoge" とわざとエラーを起こしています。 $ digdag run qiita1.dig (...略...) 2019-08-18 18:38:09 +0900 [INFO] (main): Starting a new session project id=1 work

Deleted articles cannot be recovered. Draft of this article would be also deleted. Are you sure you want to delete this article?

原因は何だったか A.dig、B.digともにdailyオプションでschedule指定をしていたため、session_timeが両者ともに00:00:00 となっていた。 requireは親となるワークフローのsession_timeで子ワークフローを起動する ため、C.digが同一のsession_timeで2回呼び出されることとなり、時系列順で後の処理が不発となっていた。 挙動をまとめてみる scheduleのパラメータについて ドキュメントにはちゃんと書いてあります。が、完全に見逃していました。 session_timeが00:00:00(または00:00)となる挙動は、dailyだけでなくhourly/weekly/monthlyオプションでも同様です。 https://docs.digdag.io/scheduling_workflow.html?highlight=minut

概要 現職で digdag を集計バッチをコントロールするジョブサーバとして利用していますが、いくつか機能的に足りないところがあったので Python で機能拡張してみました。それについて簡単に紹介したいと思います。 不足してる機能 ジョブ待ち合わせ ワークフロー require オペレータが存在し、これによって複数のワークフロー間の依存関係を記述することは既存機能で可能です。しかしながら、待ち合わせができるワークフローは同一プロジェクト内のワークフローに限られています。「依存関係があるならば同一プロジェクトで管理しろ」と言われそうですが、よんどころない理由で別プロジェクト管理にしたい場合には依存関係をもたせることができません。 タスク _parallel: true を記述して並行実行しているところで一部だけタスクの待ち合わせをしたい場合、タスクの記述順序とかを駆使して書く必要があり、若

概要 データパイプラインの管理にワークフローエンジンを導入したいのですが、今の要件に対してどれが合っているのか判断しきれない部分があるので整理してみました 最近の導入事例や発表をみるかぎりAirflow, Argo, Digdagあたりが人気なのかなと思います ワークフローエンジンとは ワークフローエンジンとは定期的なバッチ処理をうまく処理できるように、バッチ実行を管理してくれるソフトウェアのことです 古典的な実現方法としては適当なlinuxサーバーの上でcron実行させることが考えられますが、以下のような問題があります ジョブごとの依存関係を表現できない。cronの時間指定で実現させようとすると、タスクAを１時に開始してそれが完了するとみなして依存するタスクBを２時に開始するというような書き方をすることになるが、実際にタスクAが２時までに終わらなかった場合に処理が上手く実行できない タス

s3_waitについて(部分意訳・要約) https://docs.digdag.io/operators/s3_wait.html s3_waitとは s3にファイルが置かれるのを待つdigdagオペレーター オペレーター表記 s3_wait> ポイント 環境変数の登録が前提条件 ファイルが作成されてから演算子がそれを検出するのはラグがある 待機中はワークフローの実行は起動中のまま 環境変数 s3を利用するのでアクセスキーとシークレットキーは必須 前提として、下記のように登録・指定が必要 # 例 ## ローカルで実行する場合 digdag secrets --local --set aws.access_key_id='' digdag secrets --local --set aws.secret_access_key='' ## digdagサーバー上で実行する場合 digdag

Deleted articles cannot be recovered. Draft of this article would be also deleted. Are you sure you want to delete this article?

Deleted articles cannot be recovered. Draft of this article would be also deleted. Are you sure you want to delete this article?

Deleted articles cannot be recovered. Draft of this article would be also deleted. Are you sure you want to delete this article?

Deleted articles cannot be recovered. Draft of this article would be also deleted. Are you sure you want to delete this article?

Deleted articles cannot be recovered. Draft of this article would be also deleted. Are you sure you want to delete this article?

Deleted articles cannot be recovered. Draft of this article would be also deleted. Are you sure you want to delete this article?

Digdagワークフローエンジンを利用したデータ監視ワークフロー作成 この記事はオールアバウトアドベントカレンダー18日目の記事です(投稿のお休みの日もあるので、事実14日目？)。 データエンジニアとして普段からトレジャーデータやBigQueryはもちろん、既存のRDBMSも業務で使っていますが、今回はオールアバウトで提供しているアドネットワーク「プライムアド」の広告配信監視をワークフローで実装してみたので、その概要と失敗談をまとめてみました。 もちろん、アドネットワークの監視はここで紹介するワークフロー以外にも色々の監視体制をとっています。 むしろワークフローによる監視は副次的なもので、データエンジニアの観点で、データからの監視はできないか？という問いから試してみたことになります。 Digdagについて みなさん、Digdagはご存知でしょうか？オールアバウトではトレジャーデータを利用し

今までは EC2 を 1 台立てて、その中で Digdag サーバを立てて、上記全ての機能を担っていたが、それをやめて複数台の冗長構成にしたい REST API Server に関しては 1 台で、それ以外の Task Agent、Workflow Executor、Schedule Executor を行うサーバを冗長構成にする また、EC2 の AutoScalingGroup でオートスケール構成にする 割り込みタスクが即時実行されること（タスクのキュー待ちが発生しない） 現在、Digdag 上で毎時で約 450 タスクを起動するワークフローが実行されている 各々のタスクはすぐ終わるようなものではなく、キューから中々解放されない キューが溜まっている時に、割り込みタスクが入るとキューに残ったまま実行されないので、即時実行されるようにしたい 現在、Digdag サーバの起動オプションで