Lambda : s3のcsvファイルをpandasで変換し自動でDynamoへ格納する

## はじめに

簡単な内容ですが、Lambdaでのpandasの起動やcsv読み込み、DynamoDB格納の為のfloat⇒Decimal変換等、一部嵌りポイントがあったので、備忘録も兼ねて記載します。

## やりたい事

s3にcsvファイルをアップロードしたら自動的にDynamoDBへ格納する。

s3　⇒　Lambda　⇒　DynamoDB 

## DynamoDBの作成

DynamoDBを作成します。
プライマリキーは一旦インデックスにします。
ソートキーの追加で拡張できます。

![Dynamo_create.PNG](https://qiita-image-store.s3.amazonaws.com/0/257113/f72fe1f1-0a65-6888-1f0f-2309b07e18e4.png)


## Lambda関数の作成（設定）

Lambda関数を作成します。
ロールにはs3とDynamoDBへのアクセスが可能な権限をつけておきます。
リージョンはs3と合わせます。

![lambda_1.PNG](https://qiita-image-store.s3.amazonaws.com/0/257113/b105c6d0-2779-c5d5-1486-eef784bbb08c.png)


Lambda関数のトリガーを設定します。

![lambda_trriger.PNG](https://qiita-image-store.s3.amazonaws.com/0/257113/ebf37368-3a34-db81-9653-54a47e5f2ae8.png)

## Lambda関数の作成（コード）

```Python:Lambda_function.py
import boto3
import json
import io
import decimal
import pandas as pd

from decimal import *
getcontext().prec = 6

dynamodb = boto3.resource('dynamodb')

# convert function to store data to dynamodb.
def convert_float_to_Decimal(dictionary):
    for key in dictionary:
        if isinstance(dictionary[key], float):
            dictionary[key] = Decimal(str(dictionary[key]))
    return dictionary

def insert(index, dictionary):
    dictionary["index"] = int(index)
    import_item = convert_float_to_Decimal(dictionary)
    table = dynamodb.Table('test')
    table.put_item(Item=import_item)

# read csv from s3 and insert data as dictionary to dynamodb.
def lambda_handler(event, context):
    client = boto3.client('s3')
    # retrieve directory from event.
    key = event['Records'][0]['s3']['object']['key']
    obj = client.get_object(Bucket='dataset-pract',Key=key)
    initial_df = pd.read_csv(io.BytesIO(obj['Body'].read()),
    index_col=0)
    initial_df_index = list(initial_df.index)
    # rewrite df as list of dict.
    initial_df = initial_df.to_dict("records")
    for index, dictionary in zip(initial_df_index, initial_df):
        insert(index, dictionary)
```

### float型をDecimal型に変換 
DynamoDBにはfloat形式ではデータを入れることができないので、格納したい辞書形式すべてのkeyに対して、valueがfloatであればDecimalに変換しています。
変換の際、floatから直接Decimalに変換するとうまくいかないため、一旦stringを挟みます。 
 <https://github.com/boto/boto3/issues/665>

### csvの読み込み 
get_objectで指定したバケットから読み込みます。その後、Bodyを展開してバイトデータを変換し、read_csvで読み込みます。

### Pandasのimportについて 
Lambda上でPandasを使うにはパッケージをインストールする必要があります。
この場合、PandasのフォルダとLambda関数をzip化しアップロードすれば良いのですが、Pandasのインポートには依存関係があるので下記サイトからダウンロードしたファイルを使用するのが楽です。

[Lambda用Pandas](https://github.com/pbegle/aws-lambda-py3.6-pandas-numpy)

![pandas_package.PNG](https://qiita-image-store.s3.amazonaws.com/0/257113/4376657c-98a6-2579-c58d-2b1f2e743763.png)

このようにダウンロードしたフォルダに作成したLambda関数を入れてzip化します。

zipをアップロード

![lambda_zip.PNG](https://qiita-image-store.s3.amazonaws.com/0/257113/053e5554-f636-26c1-bac3-8c16c789da43.png)

トリガー起動時に実行する関数を指定します。
今回はLambda_handlerにします。

アップロードして保存を行えば完了です。

## テスト

使用するデータ

```Python:
import pandas as pd
data = pd.read_csv("USAccDeaths.csv", index_col=0)
print(data.head())
```

<img src="https://qiita-image-store.s3.amazonaws.com/0/257113/1a7a7c55-7080-bebf-a006-61bc70a77f6c.png" width=50%>

これをs3の指定したバケットに入れます。



DBを確認します。

![Dynamo.PNG](https://qiita-image-store.s3.amazonaws.com/0/257113/475c916e-98c3-dc3f-8c3b-3c98cb47c81b.png)

無事入りました。


以上です。