Lambdaで取得した画像をKinesis及びFirehose経由でS3にアップロードする

Kinesisの動作確認のため、Lambdaから画像を流した時の個人メモ

[（参考）Lambdaで取得した画像をKinesis Firehose経由でS3にアップロードする](https://qiita.com/nkmk1215/items/42702818aa9f19fb56e1)
[（参考）Kinesis — Boto 3 Docs 1.9.96 documentation - Amazon AWS](https://boto3.amazonaws.com/v1/documentation/api/1.9.42/reference/services/kinesis.html#Kinesis.Client.put_record)

---

# Kinesis
- 「データストリームの作成」を押下

![000.JPG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/242897/3f9faf87-1793-be88-f263-ecbec15cf79a.jpeg)

- シャード数は最小値として作成

![001.JPG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/242897/0b37244c-12ac-dee3-8b7d-c31bbc2781ac.jpeg)

- 作成されたことを確認

![002.JPG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/242897/54849a12-9023-46dc-3eb1-f832e1253458.jpeg)

# Kinesis Firehose
- 「配信ストリームを使用した処理」を押下し、Firehose作成画面に遷移する
    - [Lambdaで取得した画像をKinesis Firehose経由でS3にアップロードする](https://qiita.com/nkmk1215/items/42702818aa9f19fb56e1)にて  
      作成したFirehoseはインプット元として「Direct PUT」を選択したため、流用できない

![003.JPG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/242897/cfc64738-9775-b661-94f7-7b44757f1777.jpeg)

- Kinesisから転送するので「Amazon Kinesis Data Streams」とし、流す先としてS3を選択

![021.JPG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/242897/bff42968-351a-14f8-7373-3b839ed789a7.jpeg)

- 今回はS3 Buffer hintsを最小値に設定

![005.JPG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/242897/213f8084-e1dc-27ab-f321-91b807601142.jpeg)

- 他設定は[Lambdaで取得した画像をKinesis Firehose経由でS3にアップロードする](https://qiita.com/nkmk1215/items/42702818aa9f19fb56e1)を参照

- 作成されたことを確認

![007.JPG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/242897/68fa2cab-951c-1305-cb48-5e76eaa5d16c.jpeg)

## 動作確認
- コンソール画面からデモデータを送信

![008.JPG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/242897/59114ee6-3e46-c6d3-826c-325308b6faaf.jpeg)

- しばらくしたら送信を止める

![009.JPG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/242897/10145c02-3bc9-1fdf-2607-5a73a0f8103b.jpeg)

- S3に出力されていることを確認

![011.JPG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/242897/fb2d0c01-afc5-8c9d-9369-025e4e01c99d.jpeg)

- [Lambdaで取得した画像をKinesis Firehose経由でS3にアップロードする](https://qiita.com/nkmk1215/items/42702818aa9f19fb56e1)のようにLambdaからPutRecordするとエラーとなる
    - Firehose作成時に「Direct PUT」を選択していないから？

```
{
  "errorMessage": "An error occurred (InvalidArgumentException) when calling the PutRecord operation: This operation is not permitted on KinesisStreamAsSource delivery stream type.",
  "errorType": "InvalidArgumentException",
  "requestId": "e48441e7-7cb2-41e8-adea-23fba73b12b6",
  "stackTrace": [
    "  File \"/var/task/lambda_function.py\", line 54, in lambda_handler\n    result = put_to_firehose(img)\n",
    "  File \"/var/task/lambda_function.py\", line 38, in put_to_firehose\n    response = firehose.put_record(\n",
    "  File \"/var/runtime/botocore/client.py\", line 386, in _api_call\n    return self._make_api_call(operation_name, kwargs)\n",
    "  File \"/var/runtime/botocore/client.py\", line 705, in _make_api_call\n    raise error_class(parsed_response, operation_name)\n"
  ]
}
```

# Lambda
- Python3.9にて作成

![010.JPG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/242897/6c0d8d44-d379-2147-8b84-a04fa0ba23ef.jpeg)

- 自動生成されたロールを選択

![012.JPG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/242897/aa4a1d62-d17e-a22b-e947-be822f1b7717.jpeg)

- ロールにアタッチされているポリシーを以下のように修正
    - 画像の取得元であるS3へのGetObjectを付与
        - 今回は取得したバケットの別フォルダにアップロードする
    - KinesisへのPutRecordを付与
    - logsはデフォルトのまま

```json
{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "VisualEditor0",
            "Effect": "Allow",
            "Action": [
                "kinesis:PutRecord",
                "s3:GetObject",
                "logs:CreateLogGroup"
            ],
            "Resource": [
                "arn:aws:logs:（リージョン名）:（AWSアカウント）:*",
                "arn:aws:kinesis:（リージョン名）:（AWSアカウント）:stream/（ストリーム名）",
                "arn:aws:s3:::（バケット名）/*"
            ]
        },
        {
            "Sid": "VisualEditor1",
            "Effect": "Allow",
            "Action": [
                "logs:CreateLogStream",
                "logs:PutLogEvents"
            ],
            "Resource": "arn:aws:logs:（リージョン名）:（AWSアカウント）:log-group:/aws/lambda/testingKinesis:*"
        }
    ]
}
```

- ソース修正

```lambda_function.py
import os
import logging
import boto3
import json
import random	

# ログ設定
logger = logging.getLogger(__name__)
logger.setLevel(os.environ['LOG_LEVEL'])

# 環境変数から取得
BACKET_NAME = os.environ['BACKET_NAME']
FILE_PATH = os.environ['FILE_PATH']
STREAM_NAME = os.environ['STREAM_NAME']


# S3から指定したバケット、フォルダ、ファイル名の画像を取得する
# 取得できる画像サイズはLambdaのメモリーに依存
# return:画像のバイナリデータ
def get_img_from_s3():
    logger.debug('start get_img_from_s3()')

    s3 = boto3.client('s3')
    response = s3.get_object(Bucket=BACKET_NAME, Key=FILE_PATH)
    body = response['Body'].read()

    logger.debug('end get_img_from_s3()')

    return body


# 画像をKinesisに流す
# img:画像のバイナリデータ
# return:プットした結果
def put_to_kinesis(img):
    logger.debug('start put_to_kinesis()')
    
    kinesis = boto3.client('kinesis')
    response = kinesis.put_record(
        StreamName=STREAM_NAME,
        Data=img,
        PartitionKey=str(random.randint(1,4))
    )

    logger.debug('end put_to_kinesis()')

    return response


def lambda_handler(event, context):
    logger.debug('start lambda_handler()')

    img = get_img_from_s3()
    result = put_to_kinesis(img)

    logger.debug('end lambda_handler()')

    return result

```

- 環境変数設定

![014.JPG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/242897/f49c23c7-5719-a4c5-de8c-aa35d072ce91.jpeg)

# 動作確認
- [Lambdaで取得した画像をKinesis Firehose経由でS3にアップロードする](https://qiita.com/nkmk1215/items/42702818aa9f19fb56e1)と同様

- アップロードされたことを確認
![022.JPG](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/242897/4655c24d-4704-0aa3-e1f1-dda76f9d27b7.jpeg)

# 感想
- S3 Buffer hintsを最小値にしたことで1分たたずにS3にアップロードされた
    - 厳密に１分ではなかった
- 連続で画像を流した場合は１ファイルとしてまとめられる
    - 直接Firehoseに流した場合と同様