SnowflakeとDigdagの連携

# 1.What is Digdag？
Digdagは、タスクの複雑なパイプラインの構築、実行、スケジュール、監視を支援するシンプルなツールです。タスクが順番に、あるいは並行して実行されるように、自らファイルで定義してワークフローを記述します。

DAGをインフラとして使用することで、機械学習パイプライン、データ処理（そのほか）等を一元的に管理できるものになっています。

同種のサービスとして、
Apacheが提供しているAirflow
GCPではCloud Composerなどが有名です

# 2.Digdagのはじめかた
5分程度で終わるGetting startがあります。もし興味があればこちらからはじめてみてください。
簡単にテストコードが実行可能です。

https://docs.digdag.io/getting_started.html

# 3.アーキテクチャ
１番でも説明した通り、DigdagはDAGを定義することで
この処理が終了したらこの処理を行うなどの設定を行います。
例えば、
・MySQLからデータを取得してきて
・別のDBからデータを取得してきて
この２つをJOIN句を使って結合させる

Digdagでは親タスク、子タスクと分けて実行を管理します。

https://docs.digdag.io/architecture.html


# 3.実行コード（冗長バージョン）

※最初にDigdagを使う人用にぱっと見でわかりやすいように書いてます。
 もし、完成形のコードをみたい方は4番に飛んでください。

私は、今回Digdagのお試しとして、snowflakeのDB操作をDigdagで定義しました
この動作は私のプロジェクトの一部なので、これからさらにDigdagを拡張させたいと思います。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2526914/652da819-e2b6-d165-6f9b-f53cc55b6498.png)

こちらにsnowflakeへのプラグインが公開されていました。ありがたくいただくこととします＾＾

https://github.com/flinters/digdag_operator_snowflake

上の記載の通り、TempテーブルとMERGE INTOコマンドを自動化させる必要がありました。
先に、Tempファイルの自動化をおこなっていきます。

## 3.1 テーブル作成の自動化
上の処理を実際に記述していきます。
まずはテーブル作成を行なっていきます

create_temp_table_pm11.dig
```digdag
_export:
  plugin:
    repositories:
      - https://jitpack.io
    dependencies:
      - com.github.flinters:digdag_operator_snowflake:v0.1.3
    snow:
        host: hexemrw-XXXXXX.snowflakecomputing.com
        user: snowflake-user
        warehouse: compute_wh
        session_unixtime_sql_variable_name: session_time
        role: ACCOUNTADMIN


timezone: Asia/Tokyo

schedule:
  daily>: 23:00:00
  

+create_temp_task:
  +customer_task:
    snow>: create_temp_customer.sql 
    database: HANDM
    schema: public


  +articles_task:
    snow>: create_temp_articles.sql
    database: HANDM
    schema: public

```

_exportで実行に伴う設定の記述を行います。
今回は、snowflakeへのプラグイン接続、仮想ウェアハウスの設定でした
接続するサービスによってこれらの記述は異なるので、公式ドキュメントを見るなりしてみてください。

## timezone
timezoneは文字通り、実行の時間を設定を行います。
日本時間で設定する場合は、
Asia/Tokyoと記載します。

## schedule
どの間隔で何時に設定を行うかを記載します。
今回の場合は、毎日23時に実行させるためにこのように記載しております。

## +(タスク)
+はタスクと呼ばれるものになります。
記載順に実行が処理され、今回の場合は
親タスクを『create_temp_task』と定義し、
子タスクに実際に処理をするコードを記述しております。
.sqlで定義していますが、これはこの実行ファイルのカレントディレクトリにこのSQL文があります。

SQL
```sql
CREATE OR REPLACE TABLE temp_customer_table(customer_id string,
FN int,
active int,
club_mebmer_status string,
fashion_news_frequency string,
age int,
postal_code string);
```
### コードの実行
digdag run create_temp_table_pm11.dig
このようにターミナルで実行させます。

# 3.2 テーブル更新編
基本的にはタスクの書き方は変わりません。
設計通りに5時に実行させるように設定を行なっています。
```digdag
timezone: Asia/Tokyo

schedule:
  daily>: 05:00:00

_export:
  plugin:
    repositories:
      - https://jitpack.io
    dependencies:
      - com.github.flinters:digdag_operator_snowflake:v0.1.3
    snow:
        host: hexemrw-XXXXX.snowflakecomputing.com
        user: snow_user
        warehouse: compute_wh
        session_unixtime_sql_variable_name: session_time
        role: ACCOUNTADMIN

+upsert_task:
  +upsert_customer_table:
    snow>: upsert_customer_table.sql
    database: HANDM
    schema: public


  +upsert_articles_table:
    snow>: upsert_articles_table.sql
    database: HANDM
    schema: public

```
SQL文になります。
元テーブルとの結合を行なっております
```sql
MERGE INTO customer AS target
USING TEMP_CUSTOMER_TABLE AS source
ON (target.customer_id = source.customer_id)
WHEN MATCHED THEN
  UPDATE SET
    target.customer_id = source.customer_id,
    target.FN = source.FN,
    target.ACTIVE = source.ACTIVE,
    target.FASHION_NEWS_FREQUENCY = source.FASHION_NEWS_FREQUENCY,
    target.CLUB_MEMBER_STATUS = source.CLUB_MEBMER_STATUS,
    target.POSTAL_CODE = source.POSTAL_CODE
WHEN NOT MATCHED THEN
  INSERT (
    CUSTOMER_ID,
    FN,
    ACTIVE,
    CLUB_MEMBER_STATUS,
    FASHION_NEWS_FREQUENCY,
    AGE,
    POSTAL_CODE
  )
  VALUES (
    source.CUSTOMER_ID,
    source.FN,
    source.ACTIVE,
    source.CLUB_MEBMER_STATUS,
    source.FASHION_NEWS_FREQUENCY,
    source.AGE,
    source.POSTAL_CODE
  );

```

# 4.処理の実行（冗長したコードの整理）
実際に記述するコードをお見せしました。

ですが、あくまで最初にDigdagを使用する人用に書いたコードです。

### ２つのコードで共通している部分が複数存在する
共通している部分は、基本的にYAMLファイルで定義し、コードの冗長化を防ぎたいところです。

!inludeを使うことで、他ファイルを読み込むことができます。
つまり、YAMLファイルをそれで読み込むのです。

### スケジュールが異なる処理は別のファイルでまとめる
また、スケジュール時間が違うものは、別々のdigファイルでまとめます。

### call>:コマンドで実行ファイルを読み込む
別々のdigファイルを作成したら、それをまとめる実行ファイルを作成します。
その際に、call>:コマンドを用いて、digファイルを呼び込みます。
以下が冗長したコードを簡素化させたものになります。

```digdag
_export:
  !include : /Users/uchiiyusaku/Desktop/handm-analytics/digdag/snowflake_digdag_3/config.yaml

+run_task:
  +create_temp_table:
    call>: /Users/uchiiyusaku/Desktop/handm-analytics/digdag/snowflake_digdag_3/create_temp_table_pm11.dig
  +upsert_table:
    call>: /Users/uchiiyusaku/Desktop/handm-analytics/digdag/snowflake_digdag_3/upsert_task_am7.dig

```
YAMLファイルの中身は以下の通りです。
```yml
plugin:
  repositories:
    - https://jitpack.io
  dependencies:
    - com.github.flinters:digdag_operator_snowflake:v0.1.3
snow:
    host: hexemrw-XXXXXX.snowflakecomputing.com
    user: snow_user
    warehouse: compute_wh
    session_unixtime_sql_variable_name: session_time
    role: ACCOUNTADMIN
    database: HANDM
    schema: public

```
まとめたバージョンのdigファイルはこの通りとなってます。

create_temp_table_pm11.dig
```digdag
timezone: Asia/Tokyo

schedule:
  daily>: 23:00:00

+create_customer_temp_table:
  snow>: create_temp_customer.sql

+create_articles_temp_table:
  snow>: create_temp_articles.sql
```


upsert_task_am7.dig
```digdag
timezone: Asia/Tokyo

schedule:
  daily>: 07:00:00

+upsert_customer_table:
  snow>: upsert_customer_table.sql

+upsert_articles_table:
  snow>: upsert_articles_table.sql
```

SQL文は当然ですが、変更はしていません。

めちゃくちゃ綺麗になりました。

# 5.これらの展望
今回は、snowflakeの処理のみを自動化させましたが、他のクラウドサービスと連携させて、digdagを拡張させていきたいと思います。

### エラーを検知させるコードも実行していきたい
公式ドキュメントに記載されていた手法になります。こちらもどうせなら入れていきたいです。
```digdag
+example:
  sh>: your_script.sh
  _check:
    +succeed:
      echo>: success
  _error:
    +failed:
      echo>: fail
```