Blog-TryKinesisLambda

app.py

import base64
import json
import time

from datetime import datetime

def lambda_handler(event, context):
    print(datetime.now().isoformat())

    records_count = len(event['Records'])
    print(f'kinesis recourds count: {records_count}')

    for record in event['Records']:
        b64_data = record['kinesis']['data']
        data = base64.b64decode(b64_data).decode('utf-8')
        print(data)

    # 一瞬で終わらないように少しだけ待つ
    time.sleep(0.05)
    # time.sleep(5)

    return {
        'statusCode': 200,
        'body': json.dumps(
            {'message': 'hello world'}
        ),
    }

template.yaml

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: TryKinesisLambda

Globals:
  Function:
    Timeout: 10

Resources:
  TestStream:
    Type: AWS::Kinesis::Stream
    Properties:
        Name: test-stream
        ShardCount: 1

  HelloWorldFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: hello_world/
      Handler: app.lambda_handler
      Runtime: python3.6
      Events:
        HelloWorldStream:
          # https://github.com/awslabs/serverless-application-model/blob/master/versions/2016-10-31.md#kinesis
          Type: Kinesis
          Properties:
            Stream: !GetAtt TestStream.Arn
            StartingPosition: LATEST
            BatchSize: 3