運用が楽なQuickSightダッシュボードを作る_6（データセット自動更新+α編）

# はじめに
前回、ファイルが更新されたらQuickSightデータソースを自動更新する仕組みを記事にしました。


ですが完了を監視したり通知したりはできていなかったので、それらの仕組みを含んだ構成を作成しました。



# 構成

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/553680/b4d12812-f577-e045-f02b-eb91dcee7e5f.png)


大まかな流れは以下。
1. S3のファイル更新をEventBridgeで検知し、Step Functionsを実行
1. LambdaでQuickSight更新処理の前準備
1. QuickSightにデータセットの更新指示
1. 終わったかを検知
1. 終了内容を見て、SNSに通知

# コード
「共通リソース部」と「自動更新部」で分けています。

- 共通リソース部
    - 他の仕組みでも使いそうなリソース
    - CloudFormationで定義
        - 運用メール通知用SNSトピック
        - 設定値格納用DynamoDB
- 自動更新部
    - 自動更新にのみ使うリソース
    - AWS SAMで定義
        - EventBridge
        - Step Functions
        - Lambda


## 共通リソース部
プロジェクト単位の設定値格納場所とSNSを定義します。
```yaml
AWSTemplateFormatVersion: 2010-09-09
Description: Create Common Resource On low Burden Operation Dashboard.
Parameters:
  ProjectId:
    Type: String
    Default: "sampleproject"

Resources: 
  # SNS
  OparationTopic:
    Type: AWS::SNS::Topic
    Properties:
      TopicName: !Sub "${ProjectId}-operation-topic"
      Tags:
          - Key: "usefor"
            Value: !Sub "send e-mail to ${ProjectId} operator"
  
  # DynamoDB
  ParameterTable:
    Type: AWS::DynamoDB::Table
    Properties:
      TableName: !Sub "${ProjectId}-parameters"
      AttributeDefinitions:
        - AttributeName: S3uri
          AttributeType: S
      KeySchema:
        - AttributeName: S3uri
          KeyType: HASH # パーティションキー
      # option. If not specified, the default is PROVISIONED.
      BillingMode: PAY_PER_REQUEST
      Tags:
          - Key: "usefor"
            Value: !Sub "store using on ${ProjectId}"
  
Outputs:
  OparationTopic:
    Value: !Ref OparationTopic
    Export:
      Name: !Sub "${ProjectId}-operation-topic-arn"
  ParameterTable:
    Value: !Ref ParameterTable
    Export:
      Name: !Sub "${ProjectId}-parameters-tablename"
  ParameterTableArn:
    Value: !GetAtt ParameterTable.Arn
    Export:
      Name: !Sub "${ProjectId}-parameters-tablename-arn"
```

DynaomoDBは`BillingMode: PAY_PER_REQUEST`を指定し、従量課金にしています。
Outputは、自動更新部作成時に参照するために使います。

DynamoDBには
- キーとして、ファイル格納フォルダパス
- datasetsに、更新するデータセット群の情報を格納
- (任意)useforに、用途をメモ書き

というデータを登録します。
AWS CLIから登録する際のサンプルファイルは以下になります。
```yaml:SampleIngestionParameter.yaml
# aws --version が　v2 で実行可能（Cloudshellで可能、Cloud9は不可）
# aws ddb put <tablename> file://SampleIngestionParameter.yaml
- 
  S3uri: "s3://<<BUCKET_NAME>>/dashboard/data/tokyo_covid19/"
  usefor: "Create and Monitor QuickSight Ingestion."
  datasets:
    - 
      datasetid : "hogehoge"
      datasetname : "moemoe"
    - 
      datasetid : "togetoge"
      datasetname : "ageage"
```



## 自動更新部
AWS SAMで作っています。
フォルダ階層は以下のようにしています。
```
ingestion-sam
├── functions # Lambdaの場所
│   ├── check_ingestion_status
│   │   └── app.py
│   └── product_ingestion_parameters
│       └── app.py
├── statemachine # Step Functionsの場所
│   └── sfn.asl.json
└── template.yaml
```

### Step Functions

Step Functionsのワークフローは以下になります。
Lambdaは2か所で、他はStep Functionsを組み合わせて作成しました。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/553680/d94e9102-3286-d8f5-1681-7058743c988f.png)

1. 前処理Lambdaを実行
    1. eventのファイル情報を、後処理で使う形に成型して格納
    1. 更新処理に使うID用の数値列を発番して格納
1. DynaomoDBに検索
1. 検索結果の配列部分を並列にして、以下の処理を実行
    1. QuickSightのデータ処理に更新指示
        1. IDは、"{データセット名}-{発番した数値列}"
    1. 数秒待機
    1. 更新指示の完了をチェック
        1. Statusが継続中であれば、”数秒待機”に戻る
        1. Statusが終了していれば、後続処理へ
    1. Passで何もせず、並列処理部分を終了
        1. Choiceで並列処理は終了できなかったため置いたもの、加工等は無し
1. 各並列処理結果が配列で帰ってくるため、中身をチェック
1. 結果をSNSに通知
1. チェック結果によって、正常終了か異常終了かに分別

定義のコードは以下。DynamoDBの取得結果が`{"key":{"データ型":"value"}}`のように癖がありますので、その点注意。
```json:statemachine/sfn.asl.json
{
  "Comment": "A description of my state machine",
  "StartAt": "Lambda Invoke",
  "States": {
    "Lambda Invoke": {
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke",
      "OutputPath": "$.Payload",
      "Parameters": {
        "Payload.$": "$",
        "FunctionName": "${ProdParamsFunction}"
      },
      "Next": "DynamoDB GetItem"
    },
    "DynamoDB GetItem": {
      "Type": "Task",
      "Resource": "arn:aws:states:::dynamodb:getItem",
      "Parameters": {
        "TableName": "faci-parameters",
        "Key": {
          "S3uri": {
            "S.$": "$.targetPrefix"
          }
        }
      },
      "Next": "Map",
      "ResultSelector": {
        "datasets.$": "$.Item.datasets.L"
      },
      "ResultPath": "$.item"
    },
    "Map": {
      "Type": "Map",
      "InputPath": "$",
      "ItemsPath": "$.item.datasets",
      "MaxConcurrency": 0,
      "Parameters": {
        "dataset.$": "$$.Map.Item.Value",
        "ingestionidSuffix.$": "$.ingestionidSuffix"
      },
      "Iterator": {
        "StartAt": "CreateIngestion",
        "States": {
          "CreateIngestion": {
            "Type": "Task",
            "Next": "Wait",
            "Parameters": {
              "AwsAccountId": "${AwsAccountId}",
              "DataSetId.$": "$.dataset.M.datasetid.S",
              "IngestionId.$": "States.Format('{}-{}', $.dataset.M.datasetname.S,$.ingestionidSuffix)"
            },
            "Resource": "arn:aws:states:::aws-sdk:quicksight:createIngestion",
            "ResultPath": null
          },
          "Wait": {
            "Type": "Wait",
            "Next": "DescribeIngestion",
            "Seconds": 10
          },
          "DescribeIngestion": {
            "Type": "Task",
            "Parameters": {
              "AwsAccountId": "${AwsAccountId}",
              "DataSetId.$": "$.dataset.M.datasetid.S",
              "IngestionId.$": "States.Format('{}-{}', $.dataset.M.datasetname.S,$.ingestionidSuffix)"
            },
            "Resource": "arn:aws:states:::aws-sdk:quicksight:describeIngestion",
            "ResultPath": "$.ReturnDescribeIngestion",
            "Next": "Check Status"
          },
          "Check Status": {
            "Type": "Choice",
            "Choices": [
              {
                "Or": [
                  {
                    "Variable": "$.ReturnDescribeIngestion.Ingestion.IngestionStatus",
                    "StringEquals": "INITIALIZED"
                  },
                  {
                    "Variable": "$.ReturnDescribeIngestion.Ingestion.IngestionStatus",
                    "StringEquals": "QUEUED"
                  },
                  {
                    "Variable": "$.ReturnDescribeIngestion.Ingestion.IngestionStatus",
                    "StringEquals": "RUNNING"
                  }
                ],
                "Next": "Wait"
              }
            ],
            "Default": "Pass"
          },
          "Pass": {
            "Type": "Pass",
            "End": true
          }
        }
      },
      "Next": "Check ingestion status"
    },
    "Check ingestion status": {
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke",
      "Parameters": {
        "FunctionName": "${ChkIngestionStatusFunction}",
        "Payload.$": "$"
      },
      "Next": "SNS Publish"
    },
    "SNS Publish": {
      "Type": "Task",
      "Resource": "arn:aws:states:::sns:publish",
      "Parameters": {
        "TopicArn": "${TopicArn}",
        "Subject.$": "$.Payload.Subject",
        "Message.$": "$.Payload.Message"
      },
      "ResultPath": null,
      "Next": "Choice"
    },
    "Choice": {
      "Type": "Choice",
      "Choices": [
        {
          "Variable": "$.Payload.Status",
          "BooleanEquals": true,
          "Next": "Success"
        }
      ],
      "Default": "Fail"
    },
    "Success": {
      "Type": "Succeed"
    },
    "Fail": {
      "Type": "Fail"
    }
  }
}
```

`"IngestionId.$": "States.Format('{}-{}', $.dataset.M.datasetname.S,$.ingestionidSuffix)"`のところでは、組み込み関数を使ってIdを生成しています。使い方は公式に。

https://docs.aws.amazon.com/ja_jp/step-functions/latest/dg/amazon-states-language-intrinsic-functions.html

### 前処理Lambda

以下を行っています。
- eventから抽出したバケット名とファイルのkeyから、データ格納フォルダのS3URIを生成
- 更新処理用のIDとして、現在時刻を文字列化
```python:functions\product_ingestion_parameters\app.py
import re
import datetime

def lambda_handler(event, context):

  # eventから必要な情報を変数に格納
  targetBucket = event['detail']['bucket']['name']
  targetKey = event['detail']['object']['key']
  targetPrefix = re.sub('^(.+/)[^/]+$', r'\1', targetKey)
  targetS3uriPrefix = "s3://%s/%s"%(targetBucket,targetPrefix)
  
  # 現在日時でIngestionIdの末尾を作成
  rtnIngestionidSuffix = datetime.datetime.now().strftime('%Y%m%d_%H%M%S_%f')

  return {
    'targetPrefix':targetS3uriPrefix,
    'ingestionidSuffix':rtnIngestionidSuffix
  }
```

### 更新中身チェックLambda

- 以下の状態をチェックして、正常終了したかと、メール用メッセージを作成します。
    - ingestionStatusがCOMPLETEDで、スキップ行がなければ正常終了
        - スキップ行があればエラーにして、その旨メール用メッセージを生成
    - ingestionStatusがCOMPLETEDでなければエラー

```python:functions\check_ingestion_status\app.py
def lambda_handler(event, context):

  rtnStatus = True
  rtnSubject = ''
  rtnMessage = ''

  for eachDescribe in event:
    datasetName = eachDescribe['dataset']['M']['datasetname']['S']
    ingestionStatus = eachDescribe['ReturnDescribeIngestion']['Ingestion']['IngestionStatus']
    ingestionRowsDropped = eachDescribe['ReturnDescribeIngestion']['Ingestion']['RowInfo']['RowsDropped']
    
    if ingestionStatus == 'COMPLETED':
      if ingestionRowsDropped == 0:
        rtnMessage += "\n\t・データセット%sの更新は成功しました。"%(datasetName)
      else:
        rtnStatus = False
        rtnMessage += "\n\t・データセット%sは、%d行をスキップして更新しました。"%(datasetName,ingestionRowsDropped)
        
    else:
      rtnStatus = False
      rtnMessage += "\n\t・データセット%sの更新は失敗し、ステータス%sを戻しました。"%(datasetName,ingestionStatus)

  # end for

  if rtnStatus:
    rtnSubject = '[SUCCESS]QuickSight更新処理正常終了'
    rtnMessage = 'QuickSightの更新は正常終了しました。' + rtnMessage
  else:
    rtnSubject = '[ERROR]QuickSight更新処理異常終了'
    rtnMessage = 'QuickSightの更新処理に失敗しました。' + rtnMessage

  return {
    'Status': rtnStatus,
    'Subject': rtnSubject,
    'Message': rtnMessage
  }
```
### 構成テンプレート

- Step Functionsに、DynamoDBやSNSの操作権限をアタッチ
- EventBridgeの部分は、[前回](https://qiita.com/a_b_/items/05ec75a1034dbca31cf4)とほぼ同じです。

```yaml:template.yaml
AWSTemplateFormatVersion: 2010-09-09
Transform: AWS::Serverless-2016-10-31
Description: SAM Template for Lambda Function

Parameters:
  ProjectId:
    Type: String
    Default: "sampleproject"
  BucketName:
    Type: String
    Default: '<<BUCKET_NAME>>'
  TagetFolderPath:
    Type: String
    Description: use for prefix. End char MUST BE slash(/).
    Default: 'dashboard/data/'

Resources:
  ##################################################
  # Product Ingestion Parameters Function
  ##################################################
  ProdParamsFunctionLogGroup:
    Type: AWS::Logs::LogGroup
    Properties:
      LogGroupName: !Sub "/aws/lambda/${ProjectId}-product-ingestion-parameters"
      RetentionInDays: 3653

  ProdParamsFunctionRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: !Sub "for-lambda-${ProjectId}-product-ingestion-parameters"
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Principal:
              Service:
                - lambda.amazonaws.com
            Action:
              - 'sts:AssumeRole'
      Path: '/service-role/'
      Policies:
        # CloudWatch
        - PolicyName: write-cloudwatchlogs
          PolicyDocument:
            Version: 2012-10-17
            Statement:
              - Effect: Allow
                Action:
                  - 'logs:CreateLogStream'
                  - 'logs:PutLogEvents'
                Resource: !GetAtt ProdParamsFunctionLogGroup.Arn

  ProdParamsFunction:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: !Sub "${ProjectId}-product-ingestion-parameters"
      Role: !GetAtt ProdParamsFunctionRole.Arn
      CodeUri: functions/product_ingestion_parameters/
      Handler: app.lambda_handler
      Runtime: python3.9

  ##################################################
  # check ingestion status Function
  ##################################################
  ChkIngestionStatusFunctionLogGroup:
    Type: AWS::Logs::LogGroup
    Properties:
      LogGroupName: !Sub "/aws/lambda/${ProjectId}-check-ingestion-status"
      RetentionInDays: 3653

  ChkIngestionStatusFunctionRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: !Sub "for-lambda-${ProjectId}-check-ingestion-status"
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Principal:
              Service:
                - lambda.amazonaws.com
            Action:
              - 'sts:AssumeRole'
      Path: '/service-role/'
      Policies:
        # CloudWatch
        - PolicyName: write-cloudwatchlogs
          PolicyDocument:
            Version: 2012-10-17
            Statement:
              - Effect: Allow
                Action:
                  - 'logs:CreateLogStream'
                  - 'logs:PutLogEvents'
                Resource: !GetAtt ChkIngestionStatusFunctionLogGroup.Arn

  ChkIngestionStatusFunction:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: !Sub "${ProjectId}-check-ingestion-status"
      Role: !GetAtt ChkIngestionStatusFunctionRole.Arn
      CodeUri: functions/check_ingestion_status/
      Handler: app.lambda_handler
      Runtime: python3.9
      Timeout: 900

  ##################################################
  # Step Functions
  ##################################################
  StateMachineLogGroup:
    Type: AWS::Logs::LogGroup
    Properties:
      LogGroupName : !Sub "/aws/states/${ProjectId}-create-and-check-ingestion"

  StateMachine:
    Type: AWS::Serverless::StateMachine
    Properties:
      Name: !Sub "${ProjectId}-create-and-check-ingestion"
      DefinitionUri: statemachine/sfn.asl.json
      DefinitionSubstitutions:
        AwsAccountId: !Ref AWS::AccountId
        ProdParamsFunction: !Ref ProdParamsFunction
        ChkIngestionStatusFunction: !Ref ChkIngestionStatusFunction
        TopicArn:
          Fn::ImportValue:
            !Sub "${ProjectId}-operation-topic-arn"
      Role: !GetAtt StateMachineRole.Arn
      Logging:
        Level: ALL
        IncludeExecutionData: True
        Destinations:
          - CloudWatchLogsLogGroup:
              LogGroupArn: !GetAtt StateMachineLogGroup.Arn

  StateMachineRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: !Sub "for-statemachine-${ProjectId}-create-and-check-ingestion"
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Principal:
              Service:
                - states.amazonaws.com
            Action:
              - sts:AssumeRole
      Path: /
      Policies:
        # CloudWatch
        - PolicyName: write-cloudwatchlogs
          PolicyDocument:
            Version: 2012-10-17
            Statement:
              - Effect: Allow
                Action:
                  - 'logs:CreateLogDelivery'
                  - 'logs:GetLogDelivery'
                  - 'logs:UpdateLogDelivery'
                  - 'logs:DeleteLogDelivery'
                  - 'logs:ListLogDeliveries'
                  - 'logs:PutResourcePolicy'
                  - 'logs:DescribeResourcePolicies'
                  - 'logs:DescribeLogGroups'
                Resource: '*'
        # Lambda
        - PolicyName: invoke-lambdafunction
          PolicyDocument:
            Version: 2012-10-17
            Statement:
              - Effect: Allow
                Action:
                  - 'lambda:InvokeFunction'
                Resource:
                  - !GetAtt ProdParamsFunction.Arn
                  - !GetAtt ChkIngestionStatusFunction.Arn
        # DynamoDB
        - PolicyName: getitem-dynamodb
          PolicyDocument:
            Version: 2012-10-17
            Statement:
              - Effect: Allow
                Action:
                  - 'dynamodb:GetItem'
                Resource:
                  Fn::ImportValue:
                    !Sub "${ProjectId}-parameters-tablename-arn"
        # QuickSight describe ingestion
        - PolicyName: describe-ingestion
          PolicyDocument:
            Version: 2012-10-17
            Statement:
              - Effect: Allow
                Action:
                  - 'quicksight:CreateIngestion'
                  - 'quicksight:DescribeIngestion'
                Resource:
                  - !Sub "arn:aws:quicksight:*:${AWS::AccountId}:dataset/*/ingestion/*"
        # SNS
        - PolicyName: publish-snstopic
          PolicyDocument:
            Version: 2012-10-17
            Statement:
              - Effect: Allow
                Action:
                  - 'sns:Publish'
                Resource:
                  Fn::ImportValue:
                    !Sub "${ProjectId}-operation-topic-arn"

  ##################################################
  # EventBridge(event Rule)
  ##################################################
  ## Role for invoke step functions
  EventBridgeRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: !Sub 
        - "for-eventrule-${StateMachineName}"
        - { StateMachineName:  !GetAtt StateMachine.Name }
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Principal:
              Service:
                - events.amazonaws.com
            Action:
              - 'sts:AssumeRole'
      Path: '/service-role/'
      Policies:
        - PolicyName: exec-stepfunctions
          PolicyDocument:
            Version: 2012-10-17
            Statement:
              - Effect: Allow
                Action:
                  - 'states:StartExecution'
                Resource: !Ref StateMachine
  EventRulePutOnReceive:
    Type: AWS::Events::Rule
    Properties:
      Name: !Sub 
        - "for-invoke-statemachine-${StateMachineName}"
        - { StateMachineName:  !GetAtt StateMachine.Name }
      Description: !Sub 
        - "invoke ${StateMachineName} when file put or delete"
        - { StateMachineName:  !GetAtt StateMachine.Name }
      EventBusName: default
      State: ENABLED
      EventPattern:
        source:
          - "aws.s3"
        detail-type:
          - "Object Created"
          - "Object Deleted"
        detail:
          bucket:
            name:
              - !Ref BucketName
          object:
            key:
              - prefix: !Ref TagetFolderPath
      Targets:
        - Arn: !Ref StateMachine
          Id: !GetAtt StateMachine.Name
          RoleArn: !GetAtt EventBridgeRole.Arn
```

# 作り方

1. 共通リソース部をCloudFormationのソースを使って作成
1. `SampleIngestionParameter.yaml`を参考に、設定値をDynamoDBに登録
    1. DataSetIdの確認方法は、[以前の記事](https://qiita.com/a_b_/items/2adee15fc693307f95da#datasetid)で紹介しています。
1. SNSで、送信したいメールアドレスを設定
1. AWS SAMでデプロイ
    1. その際、共通リソース部で指定したProjectIdと同じ文字列を指定
    1. 他パラメータは、監視するバケットとフォルダを指定


# 使い方
対象のフォルダにファイルを置いたり削除したりするだけ。データセット群が更新され、結果のメールが届くはずです。

# おわりに
前回できなかった、完了チェックとメール送信を、Step Functionsを使って実装しました。
"ファイルを置いただけで、ダッシュボードが更新"という運用であれば、だいぶ使い勝手は良いのではないでしょうか。