---
title: Snowflakeのタスク失敗通知をAWSで実装してみた
description: Snowflakeのタスクが実行失敗した際に通知する仕組みをAmazon S3 × AWS Lambda × Amazon SNSで実装してみました。
---

[Snowflake Knowledge - INSIGHT LAB](https://knowledge.insight-lab.co.jp/snowflake)

# [Snowflakeのタスク失敗通知をAWSで実装してみた](https://knowledge.insight-lab.co.jp/snowflake/failed-task-notification-aws)

 作成者: [BI LAB編集室](https://knowledge.insight-lab.co.jp/snowflake/author/bi-lab)｜2020年12月24日

## はじめに

[以前の記事](https://knowledge.insight-lab.co.jp/snowflake/create_task)でご紹介した通り、  
Snowflakeには[タスク](https://docs.snowflake.com/ja/user-guide/tasks-intro.html)というSQLを定期実行できる便利な機能があります。

タスクを用いて、  
元データを整形→データマートを生成…  
といったデータパイプラインを組んでいる方は多いことでしょう。

しかし便利な機能である一方、  
タスクが実行失敗した際の通知（メールやSlackへの通知）を  
Snowflake単体で飛ばすことはできないのです…。**※2020年12月現在、私調べ**  
Webコンソール上でポチポチして通知設定できたらすごく便利なんですけどね。Snowflake様、機能追加待っております！！お願いします！！

 

　　さて どうしよ

 

…データベース管理者の皆さん。

迅速にリカバリ対応するためにも、  
タスクが実行失敗した際の通知は必須ですよね。

通知機能を備えたデータ連携ツール等を導入していれば問題ないかもしれませんが、  
今回はそういったツールが導入されていない（導入できない）場合を想定し、  
AWSの各サービスを組み合わせて通知の仕組みを実現してみようと思います！

 

## TASK\_HISTORY

Snowflakeは、  
タスクの実行ログを[TASK\_HISTORY](https://docs.snowflake.com/ja/sql-reference/functions/task_history.html)から確認できます。

上記SQLを実行すると…

こんな感じで結果が返ってきます。  
沢山カラムがありますね。

主要なカラムの説明は下記表を参考に。  
※全カラムの詳細な説明は公式ドキュメントの[TASK\_HISTORY](https://docs.snowflake.com/ja/sql-reference/functions/task_history.html)に記載されています！

| **カラム名** | **説明** |
| --- | --- |
| **QUERY\_ID** | **SQLステートメントのID** |
| **NAME** | **タスク名** |
| **DATABASE\_NAME** | **タスクの格納先DB名** |
| **SCHEMA\_NAME** | **タスクの格納先スキーマ名** |
| **QUERY\_TEXT** | **SQLステートメントの内容** |
| **STATE** | **タスクのステータス：SCHEDULED/EXECUTING/SUCCEEDED/FAILED/CANCELLED** |
| **ERROR\_CODE** | **エラーコード** |
| **ERROR\_MESSAGE** | **エラーメッセージ** |
| **SCHEDULED\_TIME** | **タスクのスケジュール時刻** |
| **QUERY\_START\_TIME** | **タスクの実行時刻** |
| **COMPLETED\_TIME** | **タスクの完了時刻** |

 

ふむふむ…

この実行ログから必要なデータを抽出して…定期的にどこかに置いて…  
メッセージ生成プログラムを発火⇒通知みたいな仕組みを作れば…いけそうな…気が…するぞ？

 

## ではどうするのか？

今回はAWSのサービス3つを使用します。

- [Amazon S3](https://aws.amazon.com/jp/s3/)（Simple Storage Service） 
    - みんな大好きオブジェクトストレージ
    - ファイルの置き場所
- [AWS Lambda](https://aws.amazon.com/jp/lambda/) 
    - みんな大好きサーバーレスなプログラム実行環境
    - Python/Node.js/Java/Ruby/COBOL等が動く
- [Amazon SNS](https://aws.amazon.com/jp/sns/)（Simple Notification Service）  
    - みんな大好きフルマネージドなメッセージングサービス
    - メールに簡単に飛ばせる
    - Lambda ⇒ SNS ⇒ Lambdaのように仲介役も可能

EC2とかは使いません。コスパ重視。サーバーレス最高。

 

そして、下記の構成で無理やりスマートに通知機能を実現します！

 

**①定期的にタスクの実行ログをS3にアンロード  
②S3トリガーでLambdaを起動して通知用メッセージを生成  
③SNSにメッセージをキューイング  
④メールを送信**

 

これでなんとなく通知機能のイメージはできましたか？

 

## 通知機能の要件

続いて通知機能の要件をざっくり決めちゃいます！

- **タスクが実行失敗した際、30分以内にメールで通知されること**
- **実行失敗したタスク単位でメールを送信すること** 
    - **例）30分以内に3タスク実行失敗したら3つメールを送信**
- **メールのタイトルにタスク名を含めること**
- **本文には下記の情報を含めること** 
    - **タスク名**
    - **DB名**
    - **タスク終了時刻**
    - **エラーコード**
    - **エラーメッセージ**
- **S3には過去の実行ログを残さないこと（最新分だけ残す）**

 

これらの要件を踏まえて実装を進めます。

 

## 実装の手順

いよいよ実装フェーズに移ります！

 

### 前提

実装前に、前提条件として下記作業の実施をお願いします。

- Amazon S3上でバケット作成 
    - 任意のバケット名で作成しておいて下さい。
    - タスク実行ログの出力先として使用します。
- Snowflake上でAmazon S3への[ストレージ統合](https://docs.snowflake.com/ja/user-guide/data-load-s3-config-storage-integration.html)作成 
    - 接続・認証設定のようなものです。
    - 対象バケットのAmazon リソースネーム (ARN)が必要になります。
- Snowflake上でAmazon S3への[外部ステージ](https://docs.snowflake.com/ja/user-guide/data-load-s3-create-stage.html)作成 
    - 接続後のロードやアンロードに関する更に細かい仕様を設定するものです。
    - SnowflakeからS3のバケットが参照できるようになります。

 

### 【Amazon SNS】トピック＆サブスクリプション設定

#### トピックの作成

SNSコンソールを開き、\[トピックの作成\]をクリックします。

 

\[スタンダード\]にチェックを入れ、任意の名前を入力します。

 

必要に応じてオプションを設定し、\[トピックの作成\]をクリックします。

作成したトピックの情報が表示されます。  
※ARNは後程Lambdaで使用するので、メモしておきましょう！

 

#### サブスクリプションの作成

次に\[サブスクリプションの作成\]をクリックします。

 

\[プロトコル\]は\[Eメール\]を選択し、\[エンドポイント\]に通知先のメールアドレスを入力します。

必要に応じてオプションを設定し、  
\[サブスクリプションの作成\]をクリックします。

 

すると、先ほど入力したメールアドレス宛にAWSから確認メールが届きます。

メール本文のリンク\[Confirm subscription\]をクリックします。

 

ブラウザが立ち上がり、下記ページが表示されます。

 

トピックの画面に戻ってみましょう。  
サブスクリプションが追加され、\[ステータス\]が確認済になっていますね。

 

これでSNS側の作業は完了です！

 

### 【AWS Lambda】通知文生成処理実装

#### Pythonの実装

Lambdaコンソールを開きます。

任意の関数名を入力し、ランタイムは\[Python 3.8\]を選択して関数を作成。

 

サンプルコードが表示されます。

サンプルコードは削除し、下記のコードをCopy&Pasteします。

コードの細かい説明は省きますが、  
S3のファイルをロード⇒値を変数に格納⇒通知用のタイトル&本文生成⇒SNSに送信、といった処理を組んでいます。

#### 環境変数の追加

上記のコードは環境変数からS3とSNSの接続情報を取得するようにしているため、  
環境変数を追加する必要があります。

\[環境変数の編集\]を開きます。

\[S3\_BUCKET\_NAME\]にS3のバケット名を、  
\[SNS\_TOPIC\_ARN\]にSNSのARNを入力し、保存します。

 

#### タイムアウトの設定

Lambdaはデフォルトだと3秒でタイムアウトするのですが、  
3秒だと処理が終わらない可能性があります…。念のため変更しておきましょう！

\[基本設定を編集\]を開きます。

5分あれば十分かと。  
変更したら保存します。

 

#### S3トリガーの設定

S3にファイルが出力されたタイミングでLambda関数を実行したいため、  
トリガーを設定する必要があります。

\[デザイナー\]>\[トリガーを追加\]をクリックします。

 

トリガーの一覧から\[S3\]を選択し、\[バケット\]は該当のバケット名を選択します。  
イベントタイプは\[PUT\]を選択しましょう。

※プレフィックスとサフィックスはオプションですが、バケットの構成によっては入力する必要があります。

右下の\[追加\]をクリックしたらトリガーの設定完了です。

 

\[デザイナー\]を見ると、左側にS3が追加されていると思います。

 

後はコードをデプロイしたらLambda側の作業は完了です！

 

### 【Snowflake】S3アンロード用タスク作成

Snowflake上で下記のSQLを実行します。  
ウェアハウス名と外部ステージ名は事前に作成したものを指定してください。

- 過去30分間において実行失敗(FAILED)したタスクのログを抽出し、  
  30分毎にtsv形式でS3へアンロード([COPY](https://docs.snowflake.com/ja/sql-reference/sql/copy-into-location.html))する処理を組んでいます。
- tsvファイルは同一ファイル名で上書きするようにしており、  
  S3上には最新分のファイルしか残らないようにしています。
- error\_mesageは改行コードを含む場合があるため、置換処理しています。

 

タスクは作成した直後は停止(SUSPEND)状態なので、起動(RESUME)しましょう。

これで準備が整いました！

 

## 検証

動作検証してみましょう。

検証用に、エラーで失敗するタスクをSnowflake上で作成します。

とりあえずゼロ除算エラーを起こすタスクを作成。

RESUMEして、30分ほど待ちましょうか。

 

すると…。

…。

お、通知が来たぞ？

メールボックスを開くと…

この通り！  
ちゃんとメールで通知されましたね。

 

## 余談

今回はタスク失敗通知の機能を実現すべく、  
Snowflake(TASK\_HISTORY)⇒S3⇒Lambda⇒SNSといったフローを構築しましたが、  
別にTASK\_HISTORYに限らず何のデータでもこのフローに組みことはできますので、  
例えばSnowflake上のデータマートの一部を定期的に特定メンバーへ送信…といった用途にも応用できると思います。

 

## 最後に

**この記事がいずれ不要になることを切に願います。**  
Snowflake様、機能追加待っております！！お願いします！！

以上、  
「Snowflakeのタスク失敗通知をAWSで実装してみた」でした！

[完全な記事を表示](https://knowledge.insight-lab.co.jp/snowflake/failed-task-notification-aws)

```json
{
  "@context" : "http://schema.org",
  "@type" : "BlogPosting",
  "author" : {
    "@type" : "Person",
    "name" : "BI LAB編集室"
  },
  "dateModified" : "2022-04-28T06:28:10.041Z",
  "datePublished" : "2020-12-24T17:23:40Z",
  "headline" : "Snowflakeのタスク失敗通知をAWSで実装してみた",
  "image" : {
    "@type" : "ImageObject",
    "height" : 410,
    "url" : "https://download.insight-lab.co.jp/hubfs/key-visual-1.png",
    "width" : 923
  },
  "mainEntityOfPage" : "https://knowledge.insight-lab.co.jp/snowflake/failed-task-notification-aws",
  "publisher" : {
    "@type" : "Organization",
    "logo" : {
      "@type" : "ImageObject",
      "height" : 60,
      "url" : "/hs/hsstatic/content_shared_assets/static-1.4092/img/default-amp-logo.png",
      "width" : 60
    },
    "name" : "snowflakeナレッジ"
  }
}
```