Kinesis Firehose

「Create delivery stream」を押下

Kinesisから転送はしないので「Direct PUT」とし、流す先としてS3を選択

S3バケット名を指定
- ここで指定したバケットへのアクセスを含めたIAMロールが自動生成される

他はデフォルトのまま、作成

数分後、作成されたことを確認

Lambda

Python3.9にて作成

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

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

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

ソース修正

lambda_function.py

import os
import logging
import boto3
import json

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

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


# 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


# 画像をFirehoseに流す
# img:画像のバイナリデータ
# return:プットした結果
def put_to_firehose(img):
    logger.debug('start put_to_firehose()')

    firehose = boto3.client('firehose')
    response = firehose.put_record(
        DeliveryStreamName=DELIVERY_STREAM,
        Record={
            'Data': img
        }
    )

    logger.debug('end put_to_firehose()')

    return response


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

    img = get_img_from_s3()
    result = put_to_firehose(img)

    logger.debug('end lambda_handler()')

    return result

環境変数設定

動作確認

流す画像をS3に配置

任意のテストイベントJSONを設定し、実行

Lambda実行後、5分後にS3を確認し、ファイルをダウンロード

取得したファイルに拡張子をつけ、pngファイルとして開く

感想

仕様的に流れてきたログ等をいったん溜めてS3にアップロードしている
- S3へのアップロードが5MBか300秒に達したときになっているため
  連続して流した場合1つのファイルに2つ分の画像が含まれる
- ファイル名がFirehoseによってリネームされる
  - ログ等をまとめてS3に流すのであれば
    一意のファイル名をつけてくれるので助かるが……

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

Kinesis Firehose

Lambda

動作確認

感想