概要

DigdagのDBプラグインとしてPostgreSQLやMySQLが存在します。しかし使いたいDB(今回はPureData)に対応しているプラグインがありませんでした。
そこでJDBCドライバーさえあれば使えるdigdag-myjdbcというプラグインを作ろうとしたので紹介します。
なぜ名前をmyjdbcにしたかと言うと、すでにDigdag本体の中でjdbcという名前がインターフェースとして使われていたからです。

ソースコード

ソースコードは以下のリポジトリを参照してください。

https://github.com/katsuyan/digdag-plugin-myjdbc

基本的に、Digdagにデフォルトで付いているPostgreSQLのプラグインと@hiroysatoさんが開発をしているdigdag-plugin-mysqlを参考に作成しました。

動かしてみる

それでは、実際に動かしてみましょう。
基本的には、PostgreSQLプラグインと同じインターフェースになっており、
JDBCとDB情報を外から与えてやることでDigdagからDBにアクセスすることが可能となっています。

例としてdigdag-myjdbc-pluginを利用してPostgreSQLから利用データ取得し表示をしてみます。
digdasg-myjdbc-pluginでは独自にdriver_name, protocol, driver_pathという項目を追加しています。

ソースコード

myjdbc.dig

_export:
  plugin:
    repositories:
      - file:///Users/katsuya.tajima/src/github.com/katsuyan/digdag-plugin-myjdbc/build/repo
      # - https://jitpack.io
    dependencies:
      - com.github.katsuyan:digdag-plugin-myjdbc:0.1.1

  myjdbc:
    host: 127.0.0.1
    port: 5432
    user: digdag
    database: tmp
    driver_name: org.postgresql.Driver
    protocol: postgresql
    driver_path: postgresql-42.2.6.jar

+step1:
  myjdbc>: test.sql
  store_last_results: true

+step2:
  sh>: echo ${myjdbc.last_results}

test.sql

select * from name;

結果

$ digdag run myjdbc.dig
2019-12-06 17:34:36 +0900: Digdag v0.9.39
2019-12-06 17:34:37 +0900 [INFO] (main): Database migration started
2019-12-06 17:34:37 +0900 [INFO] (main): Applying database migration:20151204221156
2019-12-06 17:34:37 +0900 [INFO] (main): Applying database migration:20160602123456
2019-12-06 17:34:37 +0900 [INFO] (main): Applying database migration:20160602184025
2019-12-06 17:34:37 +0900 [INFO] (main): Applying database migration:20160610154832
2019-12-06 17:34:37 +0900 [INFO] (main): Applying database migration:20160623123456
2019-12-06 17:34:37 +0900 [INFO] (main): Applying database migration:20160719172538
2019-12-06 17:34:37 +0900 [INFO] (main): Applying database migration:20160817123456
2019-12-06 17:34:37 +0900 [INFO] (main): Applying database migration:20160818043815
2019-12-06 17:34:37 +0900 [INFO] (main): Applying database migration:20160818220026
2019-12-06 17:34:37 +0900 [INFO] (main): Applying database migration:20160908175551
2019-12-06 17:34:37 +0900 [INFO] (main): Applying database migration:20160926123456
2019-12-06 17:34:37 +0900 [INFO] (main): Applying database migration:20160928203753
2019-12-06 17:34:37 +0900 [INFO] (main): Applying database migration:20161005225356
2019-12-06 17:34:37 +0900 [INFO] (main): Applying database migration:20161028112233
2019-12-06 17:34:37 +0900 [INFO] (main): Applying database migration:20161110112233
2019-12-06 17:34:37 +0900 [INFO] (main): Applying database migration:20161209001857
2019-12-06 17:34:37 +0900 [INFO] (main): Applying database migration:20170116082921
2019-12-06 17:34:37 +0900 [INFO] (main): Applying database migration:20170116090744
2019-12-06 17:34:37 +0900 [INFO] (main): Applying database migration:20170223220127
2019-12-06 17:34:37 +0900 [INFO] (main): Applying database migration:20190318175338
2019-12-06 17:34:37 +0900 [INFO] (main): Database migration successfully finished.
2019-12-06 17:34:37 +0900 [WARN] (main): Reusing the last session time 2019-12-06T00:00:00+00:00.
2019-12-06 17:34:37 +0900 [INFO] (main): Using session /Users/katsuya.tajima/src/github.com/katsuyan/digdag-example/myjdbc-plugin/.digdag/status/20191206T000000+0000.
2019-12-06 17:34:37 +0900 [INFO] (main): Starting a new session project id=1 workflow name=myjdbc session_time=2019-12-06T00:00:00+00:00
2019-12-06 17:34:38 +0900 [INFO] (0018@[0:default]+myjdbc+step1): myjdbc>: test.sql
2019-12-06 17:34:39 +0900 [INFO] (0018@[0:default]+myjdbc+step2): sh>: echo [{"name":"tajima","id":1},{"name":"shiozaki","id":2}]
[name:tajima,name:shiozaki] [name:tajima,id:2] [id:1,name:shiozaki] [id:1,id:2]
Success. Task state is saved at /Users/katsuya.tajima/src/github.com/katsuyan/digdag-example/myjdbc-plugin/.digdag/status/20191206T000000+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.

対応していないこと

PostgreSQLのプラグインと同じインターフェースで使えるように意識をしたところ、汎用的にするのが難しいことがわかりました。
ここでは、このプラグインで対応していないことを原因をベースに紹介します。

SET TRANSACTION READ ONLY

jdbcのプラグインでstore_last_resugit を使う場合READ ONLYなクエリとしてDBにアクセスします。PostgreSQLのプラグインでは、書き換えが発生しないことを担保するためにSET TRANSACTION READ ONLY を実行することでクエリをラップしています。

よってSET TRANSACTION READ ONLY が使えないデータベースではエラーが発生してしまいます。実際PureDataでは BEGIN を SET TRANSACTION READ ONLY の前に実行しないとエラーが発生してしまいました。

CREATE TABLE IF NOT EXISTS

create_table オプションを利用することで、Digdag上からDBに対して CREATE TABLE 文を発行します。実際の処理は以下の行になります。

見てわかるように CREATE TABLE IF NOT EXISTS が使われています。よって、 IF NOT EXISTS に対応していないDBではエラーとなってしまいます。同じことを実現するのに汎用的なクエリはないかを探したのですが見つかりませんでした。

まとめ

やはり、DBによって対応している文法が異なるためPostgreSQLのプラグインと同じインターフェースで汎用的なプラグインを作成するのは難しかったです。
上で見たものの他にもPostgreSQLのプラグインでは CURRENT_TIMESTAMP や now() なども使われているので引っかかるDBはありそうです。

よって現状ではDB毎にプラグインを作ってやるのがベストなのかなと思いました。
それか、より汎用的なインターフェースにしたJDBCプラグインを作るのがいいかなと思います。機会があれば作ろうかなと思います...

緊急でどうしても任意のJDBCドライバーを使いたい場合はこちらをクローンして、クエリを編集して使ってください。

明日はDigdagネタ最終日で「使われていないEC2上のDigdagワーカーを自動でTerminateする」になります。