RedashでSendGridのActivityを見る(Amazon S3 -> Athena経由)

2017/12/03 2017/12/10

Redash Advent Calendar 2017に参加しましたので、今年リリースの v2 でダイレクトでの接続が実装されたAmazon Athenaデータソースを使ってSendGridのActivityを見たことについて書きたいと思います。
v2より前はDockerを使ったProxy経由でAmazon Athenaデータソースに接続していました。

概要

SenGridのEvent WebHookでAmazon API GatewayへEventを送信
Amazon API GatewayをトリガーにしたAWS LambdがEventを受け取ってCSV形式でAmazon S3へ格納
Amazon S3 をデータソースにしてAmazon Athenaでテーブルを作成
RedashでAmazon Athenをデータソースにしてクエリ実行、可視化

Amazon S3でCSVを格納するバケットを作成する

バケット名とリージョンだけ指定してあとは何も設定してません。

バケット名は「sendgrid.activity.yamamugi.com」、
リージョンはRedashのEC2を起動しているバージ二ア北部にしました。

AWS LambdaでSendGridのActivityをCSVに変換してS3バケットに格納する

ソースコードはこちらGitHubです。

def lambda_handler(event, context):
    try:
        s3_bucket = boto3.resource('s3').Bucket(bucket_name)

        body = ''
        for e in event:
            rows = []
            rows.append(get_value(e, 'email'))
            rows.append(get_value(e, 'timestamp'))
            rows.append(get_value(e, 'ip'))
            rows.append(get_value(e, 'sg_event_id'))
            rows.append(get_value(e, 'sg_message_id'))
            rows.append(get_value(e, 'useragent'))
            rows.append(get_value(e, 'event'))
            rows.append(get_value(e, 'response'))
            rows.append(get_value(e, 'tls'))

            body += ','.join(rows)
            body += '\n'


        key_name = '{time}{id}.csv'.format(
            time=datetime.now().strftime('%Y/%m/%d/%H/%M%S'),
            id=event[0].get('sg_event_id', '')
        )

        s3_bucket.put_object(
            Key=key_name,
            Body=body,
            ContentType='text/csv'
        )

    except:
        logger.error(traceback.format_exc())
        send_message(traceback.format_exc(), slack_channel)

def lambda_handler(event, context):

try:

s3_bucket = boto3.resource('s3').Bucket(bucket_name)

body = ''

for e in event:

rows = []

rows.append(get_value(e, 'email'))

rows.append(get_value(e, 'timestamp'))

rows.append(get_value(e, 'ip'))

rows.append(get_value(e, 'sg_event_id'))

rows.append(get_value(e, 'sg_message_id'))

rows.append(get_value(e, 'useragent'))

rows.append(get_value(e, 'event'))

rows.append(get_value(e, 'response'))

rows.append(get_value(e, 'tls'))

body += ','.join(rows)

body += '\n'

key_name = '{time}{id}.csv'.format(

time=datetime.now().strftime('%Y/%m/%d/%H/%M%S'),

id=event[0].get('sg_event_id', '')

)

s3_bucket.put_object(

Key=key_name,

Body=body,

ContentType='text/csv'

)

except:

logger.error(traceback.format_exc())

send_message(traceback.format_exc(), slack_channel)

S3のオブジェクトキーは年月日時間で階層のようなプレフィックスにしています。
大量にActivityが発生するような場合はキーをランダムにしたほうがいいと思います。
このあとのクエリーではオブジェクトキーは何であっても問題はありません。

環境変数でバケット名とSlackのURL,チャンネルを指定しています。
Slackはエラー通知用にしているだけです。

Lambdaの権限

Lambdaのロールには、作成したS3バケットにput objectを許可あしたポリシーを割り当てたロールを設定しました。

Amazon API Gatewayを設定する

作成したAWS LambdaがPOSTメソッドをトリガーに起動するようにAPI Gatewayを設定しました。
Lambda関数を指定する以外はデフォルトのままでデプロイしました。

SendgridのWebHookを設定する

SendGridのダッシュボード [Settings]-[Mail Settings]で、[Event Notification]をONにして[HTTP POST URL]にAmazon API GatewayでデプロイしたAPIのエンドポイントを設定します。

ここまでの設定で、SendGridで何かイベントがおこるとS3に格納されるようになりました。

Amazon Athenaの設定

次にAmazon S3に格納されたCSVからAmazomn Athenaのテーブルを作成します。

Amazon AthenaのQuery Editorで以下のSQLを実行しました。

[add table]からGUIでテーブル作成を実行することも出来ます。

CREATE EXTERNAL TABLE sendgird.activity (
  `email` string ,
  `timestamp` int,
  `event` string,
  `useragent` string
 )
ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'
WITH SERDEPROPERTIES (
  'serialization.format' = ',',
  'field.delim' = ','
) LOCATION 's3://sendgrid.activity.yamamugi.com/'
TBLPROPERTIES ('has_encrypted_data'='true');

CREATE EXTERNAL TABLE sendgird.activity (

`email` string ,

`timestamp` int,

`event` string,

`useragent` string

)

ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'

WITH SERDEPROPERTIES (

'serialization.format' = ',',

'field.delim' = ','

) LOCATION 's3://sendgrid.activity.yamamugi.com/'

TBLPROPERTIES ('has_encrypted_data'='true');

プレビューをしてみると、S3に格納されたActivityデータが見えます。

RedashでAthenaのテーブルのデータを可視化する

さて、いよいよRedashです。
バージョンは、もちろんv2以上です。

公式のRedash Amazon Athena Setupを参照して設定しました。

AWS IAMユーザーの作成

ユーザーを作成してポリシーを割り当ててアクセスキーを発行します。

ドキュメントにもありますがポリシーは下記です。
* Athenaのクエリ結果が入っているバケット配下へのs3:GetObject
* Athenaのクエリ結果が入っているバケットヘのs3:GetBucketLocation , s3:ListBucket
* AWSQuicksightAthenaAccess
* AmazonAthenaFullAccess