CloudFormationでAuroraとLambdaを構築。LambdaでLOAD DATA FROM S3コマンドを実行しS3のCSVファイルをAuroraにインポートする。

本記事の内容

• AWSが提供するCloudFormationサービスを利用し、以下を行う。
  • データベースの一つであるAurora(MySQLエンジンタイプ)をシングルインスタンスで建てる。
  • サーバレスアプリケーションであるLambdaを建てる。
• 構築したAuroraに対してテーブル作成とデータ投入（LOAD DATA FROM S3）をLambdaを使って実行する。

本記事の手順で以下のような構成のクラウド環境を作成します。

※AWSの操作画面はサービスリリースと共に更新されていくため、本記事をご覧いただくタイミングによっては、現在の画面と異なる場合があります。ご了承ください。本記事内の画面キャプチャーは2022年10月時点。

DBへのテーブル作成・データ投入の方法として、EC2インスタンスからmysqlコマンドでDBにて接続して、操作を行う方法もありますが、わざわざEC2インスタンスを建てるのも面倒でしたので、Lambdaを建てそこから実施する方法を調査し実現しました。
テーブル作成はCREATE TABLEコマンドで行い、データの投入は、LOAD DATA FROM S3コマンドを実行し、S3に配置したCSVファイルをAuroraにインポートしています。

前回の記事「検証目的のためにCloudFormationでシングルインスタンスのAuroraを建てようとしたら手こずった話 」で、Auroraをシングルインスタンスで建てる説明をしましたが、その続きの話となります。

実施環境

• Windows 10
• Chrome
• AWS

構築の流れ

1. S3にLambdaデプロイ用のzipファイルとデータ投入用のCSVファイルの配置
2. CloudFormationテンプレートの準備
3. CloudFormationスタックの作成
4. Lambadaによるテストデータの投入

1. S3にLambdaデプロイ用のzipファイルとデータ投入用のCSVファイルの配置

今回のCloudFormationでのLambdaのデプロイは、事前にデプロイ用ファイルをS3に配置しておき、それをデプロイする方式をとる。
また、データ投入用のCSVファイルも配置しておく。

Lambdaにデプロイするソースコードはこちら

index.py

import boto3
import json
import pymysql.cursors
import os

def lambda_handler(event, context):
  
  #Lambdaの環境変数は、CloudFormationのデプロイ時に自動で設定する
  DB_NAME = os.environ["DB_NAME"]
  DB_USER = os.environ["DB_USER"]
  DB_PASSWORD = os.environ["DB_PASSWORD"]
  DB_WRITER_ENDPOINT_ADDRESS = os.environ["DB_WRITER_ENDPOINT_ADDRESS"]

  #pymysqlを利用してAuroraに接続するため、PyMySQLライブラリーを合わせてデプロイする必要がある
  conn = pymysql.connect(host=DB_WRITER_ENDPOINT_ADDRESS, user=DB_USER, passwd=DB_PASSWORD, db=DB_NAME, connect_timeout=5)
  with conn.cursor() as cur:

    #CreateTableコマンドは、mysqlの通常のコマンドと同じ。
    cur.execute("create table service (system_id varchar(255),name varchar(255),url varchar(2048),PRIMARY KEY (system_id))")

    #S3に配置したS3ファイルをインポートするためには"LOAD DATA FROM S3 FILE"コマンドを利用する。
    # 's3://bucket-name/testData.csv'でインポートするファイルのARNを指定しています。
    # FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n' IGNORE 1 LINES の指定で、カンマ区切り・改行が\n・先頭行は無視を指定しています。
    # (system_id,name)でインポートするカラムを指定しています。
    #cur.execute("LOAD DATA FROM S3 FILE 's3://bucket-name/testData.csv' INTO TABLE service FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n' IGNORE 1 LINES (system_id,name);")

    #こちらはSQLの従来通りのinsert intoコマンド。一つずつデータを入れる場合はこちらのSQLで良い。
    #cur.execute('insert into service (system_id, name) values("1", "test")')
    conn.commit()
    
    #インポートした結果はこちらのselect文で確認する。
    cur.execute("select * from service")
    for row in cur:
      print(row)
  
  return {
    'statusCode': 200
  }

pythonでmysqlコマンドを実行するためには、PyMySQLライブラリーが必要になるためダウンロードする。

tar.gzファイルを解凍し表示された、pymysqlフォルダーとindex.pyファイルをzipファイルに圧縮しデプロイ用ファイルを作成する。

配置用のS3バケットを作成し、Zipファイルにしたデプロイ用ファイルをS3に配置する。

これでデプロイ用ファイルが配置できた。データ投入用のCSVファイルも同様の手順で同一バケット内に配置する。
データ用ファイル「testData.csv」の中身は以下の通り。

2. CloudFormationスタックの準備

以下が実際に投入したファイルです。

aurora_template_for_Qita_Single_Instance.txt

AWSTemplateFormatVersion: 2010-09-09
#####################################################
# Parameters
#####################################################
Parameters:
  DBMasterUserName:
    Description: Please enter the name of the master user on an RDS.
    Type: String
    Default: master

  DBMasterUserPassword:
    Description: Please enter the password of the master user on an RDS. That should be greater or equal to 8 characters.
    Type: String
    Default: testdbpass
    NoEcho: true

  TestDataS3BucketArn:
    Description: Please enter the ARN of the S3 bucket where the files to be imported into the Aurora database are stored.
    Type: String
    Default: arn:aws:s3:::bucket-name

  S3BucketName:
    Description: Please enter the S3 bucket name.
    Type: String

  VpcSubnet:
    Description: AP VPC subnet
    Type: String
    Default: 10.0.0.0/16

#####################################################
# Mappings
#####################################################
Mappings:
  Constant:
    RDS:
      ClusterName: test-aurora-cluster
      DatabaseName: auroradb
      InstanceName: test-aurora-instance1
      InstanceType: db.t3.small

#####################################################
# Resource
#####################################################

#####################################################
#  VPC section
#####################################################

Description: Create VPC
Resources:
  TestVPC:
    Type: AWS::EC2::VPC
    Properties:
      CidrBlock: !Ref VpcSubnet
      Tags:
        - Key: Name
          Value: TestVPC

#####################################################
#  Subnet section
#####################################################

  TestPrivateSub1a:
    Type: AWS::EC2::Subnet
    Properties:
      CidrBlock: !Select [0, !Cidr [!GetAtt TestVPC.CidrBlock, 2, 8]] #10.0.0.0/24
      MapPublicIpOnLaunch: false
      VpcId: !Ref TestVPC
      AvailabilityZone: ap-northeast-1a
      Tags:
        - Key: Name
          Value: TestPrivateSub1a

  TestPrivateSub1c:
    Type: AWS::EC2::Subnet
    Properties:
      CidrBlock: !Select [1, !Cidr [!GetAtt TestVPC.CidrBlock, 2, 8]] #10.0.1.0/24
      MapPublicIpOnLaunch: false
      VpcId: !Ref TestVPC
      AvailabilityZone: ap-northeast-1c
      Tags:
        - Key: Name
          Value: TestPrivateSub1c

#####################################################
#  SecurityGroup section
#####################################################

  TestSecurityGroup:
    Type: AWS::EC2::SecurityGroup
    Properties:
      VpcId: !Ref "TestVPC"
      GroupDescription: Allow RDS Connection From TCP 3306 Port
      SecurityGroupIngress:
        - IpProtocol: tcp
          FromPort: 3306
          ToPort: 3306
          CidrIp: !GetAtt "TestVPC.CidrBlock"
      Tags:
        - Key: Name
          Value: TestSecurityGroup

  TestSecurityGroupForLambda:
    Type: AWS::EC2::SecurityGroup
    Properties:
      VpcId: !Ref "TestVPC"
      GroupDescription: Allow MySQL (TCP3306)
      Tags:
        - Key: Name
          Value: TestSecurityGroupForLambda

#####################################################
#   DBSubnetGroup section
#####################################################

  DBSubnetGroup:
    Type: AWS::RDS::DBSubnetGroup
    Properties:
      DBSubnetGroupDescription: RDS subnet group.
      SubnetIds:
        - !Ref "TestPrivateSub1a"
        - !Ref "TestPrivateSub1c"

#####################################################
# Aurora section
#####################################################
  #AuroraがS3にアクセスするために必要なIAMロールの設定
  RDSS3AccessRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Principal:
              Service: rds.amazonaws.com
            Action: sts:AssumeRole
      Policies:
        - PolicyName: AllowS3Access
          PolicyDocument:
            Version: 2012-10-17
            Statement:
              - Effect: Allow
                Action:
                  - s3:GetObject
                  - s3:GetObjectVersion
                Resource:
                  - !Sub '${TestDataS3BucketArn}'
                  - !Sub '${TestDataS3BucketArn}/*'

  # https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-rds-dbparametergroup.html
  DBParameterGroup:
    Type: "AWS::RDS::DBParameterGroup"
    Properties:
      Description: "RDS DB parameter group the Aurora Cluster's instance(s)."
      Family: "aurora-mysql5.7"
      Parameters:
        max_connections: "10"
      Tags: 
        - Key: Name
          Value: TestClusterParamaterGroup

  # https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-rds-dbclusterparametergroup.html
  DBClusterParameterGroup:
    Type: "AWS::RDS::DBClusterParameterGroup"
    Properties:
      Description: "RDS DB cluster parameter group for Hub-Amber"
      Family: "aurora-mysql5.7"
      Parameters:
        time_zone: 'Asia/Tokyo'
        aws_default_s3_role: !GetAtt RDSS3AccessRole.Arn #LOAD DATA FROM S3コマンドを実行し、S3のファイルを読み込むためのIAM role。超重要。
        character_set_client: "utf8mb4"
        character_set_connection: "utf8mb4"
        character_set_database: "utf8mb4"
        character_set_filesystem: "utf8mb4"
        character_set_results: "utf8mb4"
        character_set_server: "utf8mb4"
        collation_connection: "utf8mb4_bin"
        collation_server: "utf8mb4_bin"
        time_zone: "Asia/Tokyo"
      Tags: 
        - Key: Name
          Value: TestClusterParamaterGroup

  DBCluster:
    Type: AWS::RDS::DBCluster
    Properties:
      AssociatedRoles:
        - RoleArn: !GetAtt RDSS3AccessRole.Arn  #LOAD DATA FROM S3コマンドを実行し、S3のファイルを読み込むためのIAM role。超重要。
      Engine: aurora-mysql
      EngineVersion: 5.7.mysql_aurora.2.10.2
      DatabaseName: !FindInMap [Constant, RDS, DatabaseName]
      DBClusterIdentifier: !FindInMap [Constant, RDS, ClusterName]
      MasterUsername: !Ref "DBMasterUserName"
      MasterUserPassword: !Ref "DBMasterUserPassword"
      DBSubnetGroupName: !Ref "DBSubnetGroup"
      DBClusterParameterGroupName: !Ref "DBClusterParameterGroup"
      VpcSecurityGroupIds:
        - !Ref "TestSecurityGroup"

  DBInstance:
    Type: AWS::RDS::DBInstance
    Properties:
      Engine: aurora-mysql
      EngineVersion: 5.7.mysql_aurora.2.10.2
      DBInstanceClass: !FindInMap [Constant, RDS, InstanceType]
      DBSubnetGroupName: !Ref "DBSubnetGroup"
      DBParameterGroupName: !Ref "DBParameterGroup"
      DBClusterIdentifier: !Ref "DBCluster"
      AvailabilityZone: ap-northeast-1a


#####################################################
# Lambda section (データ投入用Lambda)
#####################################################
  #Lambda実行権限とAuroraの操作権限を付与したIAMロールの設定
  IAMRoleForLambdaFunction:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          -
            Effect: "Allow"
            Principal:
              Service:
                - "lambda.amazonaws.com"
            Action:
              - "sts:AssumeRole"
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/service-role/AWSLambdaVPCAccessExecutionRole
        - arn:aws:iam::aws:policy/AmazonRDSFullAccess
      MaxSessionDuration: 3600
      Path: "/"
      RoleName: "RDSFullAccessLambdaExecRole"


  LambdaFunction:
    Type: AWS::Lambda::Function
    Properties:
      Environment:
        Variables:
          DB_PORT: '3306'
          DB_NAME: !FindInMap [Constant, RDS, DatabaseName]
          DB_PASSWORD: !Ref "DBMasterUserPassword"
          DB_WRITER_ENDPOINT_ADDRESS: !GetAtt DBCluster.Endpoint.Address #Auroraのライターエンドポイントを設定している
          DB_USER: !Ref "DBMasterUserName"

      Code:
        S3Bucket: !Ref "S3BucketName"
        S3Key: index.zip
      FunctionName: "TestDataInsertLambda"
      Handler: index.lambda_handler
      Runtime: python3.8
      Role: !GetAtt IAMRoleForLambdaFunction.Arn #Lambda実行権限とAuroraの操作権限を付与したIAM
      Timeout: 10
      VpcConfig:
        SecurityGroupIds:
          - !Ref TestSecurityGroupForLambda
        SubnetIds:
          - !Ref TestPrivateSub1a

注意
VPCLambdaを作成する際にLambdaのIAMロールのポリシーに「AWSLambdaVPCAccessExecutionRole」を付与する必要がある。
NetworkInterfaceを作成する権限が必要なため、普段使用する「AWSLambdaBasicExecutionRole」だけ付与しているとCloudFormatinで以下のエラーが出る。

Resource handler returned message: "The provided execution role does not have permissions to call CreateNetworkInterface on EC2 (Service: Lambda, Status Code: 400, Request ID: 1c092832-880d-48f1-b120-1c9a364552ce)" (RequestToken: 4f51ff46-41bc-1815-0f01-69469740c1c5, HandlerErrorCode: InvalidRequest)

3. CloudFormationスタックの作成

AWSコンソールでのCloudFormationテンプレートの投入方法は、
ほぼ前回の記事通り「検証目的のためにCloudFormationでシングルインスタンスのAuroraを建てようとしたら手こずった話 」なのでそちらのキャプチャーを参考に進めてください。

以下のパラメータ入力画面が異なるため注意。

[パラメータの解説]
DBMasterUserName：Aurora Databaseの管理者アカウント名
DBMasterUserPassword：管理者アカウントのパスワード
S3BucketName：事前にLambdaのデプロイファイルを配置したバケット名 (例) cf-template-20221014-tmp ※
TestDataS3BucketArn：Auroraに投入するデータを配置しているバケットのARN (例)arn:aws:s3:::cf-test-lambda-2022
VpcSubnet：作成するVPCのサブネット

4. Lambadaによるテストデータの投入

Cloudformationにより構築したLambdaを選択しコードタブを開く。
S3からCSVファイルをインポートする場合は、以下の行のコメント外し、s3://～のtestDataファイルのS3 URIの記載を正しい値に修正する。
cur.execute("LOAD DATA FROM S3 FILE 's3://bucket-name/testData.csv' INTO TABLE service FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n' IGNORE 1 LINES (system_id,name);")

次にTestボタンをクリックし、テストイベントを設定する。イベント名はtest等で良い。引数を利用するわけでもないので、イベントJSONはそのままでよい。保存を押す。

この状態で「Test」ボタンを押下し、プログラムを実行する。
以下のような画面になり、プログラムがエラーなく終了し、DBにS3からインポートした値がselect文の結果として表示されていれば問題なく完了です。

本記事は以上です。
クラウド環境の作成の手助けになれば嬉しいです。

参考文献

本記事の作成に当たり、以下の情報も参考にさせて頂きました。ありがとうございました。

• 【AWS】実例で学ぶCloudFormation～RDS Aurora編～
• CloudFormation の Fn::Cidr 組み込み関数の使い方のメモ。
• 【AWS初学者向け・図解】CloudFormationの組み込み関数を現役エンジニアがわかりやすく解説②
• 【CloudFormation】Lambdaの Role を定義する際は !GetAtt を使って明示的に指定する
• AWS LambdaをVPC設定したときに「The provided execution role does not have permissions to call CreateNetworkInterface on EC2」
• CloudFormationでLambdaを作成する3パータン(S3/インライン/コンテナ)
• RDS（postgres）のデータをS3にexportする〜private network編〜
• VPCエンドポイントを使ってプライベートサブネットからS3へアクセスする。
• AssociatedRolesを使ってCFnでS3とデータをロード＆アンロードできるAurora MySQL5.7を構築する
• Amazon S3 バケットのテキストファイルから Amazon Aurora MySQL DB クラスターへのデータのロード
• Amazon Aurora MySQLとS3間でデータをロード&アンロードする