Kinesis Data Analytics入門　Lambdaで生成したストリーミングデータをリアルタイム分析

Kinesis Data Analyticsでストリーミングデータをリアルタイム分析する

Kinesis Data Analyticsの入門ということで、AWS公式で紹介されている構成を、実際に構築して検証します。

構築する環境

Diagram of using Kinesis Data Analytics to analyze streaming data in real time.

分析に使用するストリーミングデータは、Lambda関数で生成します。関数で実行するスクリプトは、先述の公式サイトで紹介されているPythonコードをベースにして作成します。

スクリプトは、株価をイメージしたデータを生成します。Lambda関数で生成したデータを、Kinesis Data Streamでキャプチャします。Kinesis Data Streamでキャプチャしたデータを、Kinesis Data Analyticsでリアルタイムにフィルタリングします。フィルタリングはSQLベースで実施します。Kinesis Data Analyticsは、SQLでストリーミングデータで操作することができます。具体的な検証のシナリオですが、Lambda関数で生成する株価データは、以下の5つのティッカー用のデータを対象とします。

AAPL
AMZN
MSFT
INTC
TBV

この中から1つ選び、ランダムに株価を意味する数値を生成します。今回は上記データを100個分生成し、その中からKinesis Data Analyticsで、「AAPL」のデータのみを抽出することを目標します。

CloudFormationテンプレートファイル

上記の構成をCloudFormationで構築します。以下のURLにCloudFormationテンプレートファイルを設置しています。

https://github.com/awstut-an-r/awstut-saa/tree/main/04/001

テンプレートファイルのポイント解説

今回のアーキテクチャを構成するための、各テンプレートファイルのポイントを取り上げます。

Lambda関数でKinesis検証用データを生成する

saa-04-001-lambda.yamlでLambda関係のリソースを作成します。

Resources:
  DataSourceLambda:
    Type: AWS::Lambda::Function
    Properties:
      Environment:
        Variables:
          KINESIS_STREAM_NAME: !Ref KinesisDataStreamName
      Runtime: python3.8
      Role: !GetAtt DataSourceLambdaRole.Arn
      Timeout: !Ref Timeout
      Handler: index.lambda_handler
      Code:
        ZipFile: |
          ...
Code language: YAML (yaml)

Environmentプロパティにて、関数に渡す環境変数を設定することができます。今回は後述のKinesis Data Streamsのストリーム名を渡します。Codeプロパティにて、関数内で実行するスクリプトを設定します。今回はZipFileプロパティに、直接コードを記述します。Lmabda関数の詳細については、以下のページをご確認ください。

インラインにて、株価データを生成する関数を定義します。今回は100個分の株価データを用意しますが、この関数はその1つ分のデータを生成します。

def get_data():
  return {
    'EVENT_TIME': datetime.datetime.now().isoformat(),
    'TICKER': random.choice(['AAPL', 'AMZN', 'MSFT', 'INTC', 'TBV']),
    'PRICE': round(random.random() * 100, 2)}
Code language: Python (python)

データは以下の3つの情報で構成されます。

EVENT_TIME：現在日時。
TICKER：銘柄のティッカー。ランダムに5銘柄から1つ選ぶ。
PRICE：株価。ランダムに生成する。

上記の関数を呼び出して、順次、Kinesis Data Streamに配信します。

STREAM_NAME = os.environ['KINESIS_STREAM_NAME']
LIMIT = 100

def generate(stream_name, kinesis_client, limit):
  for i in range(limit):
    data = get_data()
    print(data)
    kinesis_client.put_record(
      StreamName=stream_name,
      Data=json.dumps(data).encode('utf-8'),
      PartitionKey="partitionkey")

def lambda_handler(event, context):
  generate(STREAM_NAME, boto3.client('kinesis'), LIMIT)
Code language: Python (python)

put_recordメソッドを呼び出してKinesis Data Streamsに配信します。引数として、配信先のストリーム名、JSON化した株価データ、パーティションキーを渡します。ストリーム名は環境変数を参照して取得します。

ポイントはパーティションキーです。パーティションキーとは、配信先のシャードを決定するためのパラメータです。

パーティションキーは、ストリーム内のデータをシャード単位でグループ化するために使用されます。Kinesis Data Streams は、ストリームに属するデータレコードを複数のシャードに配分します。この際、各データレコードに関連付けられたパーティションキーを使用して、配分先のシャードを決定します。
Amazon Kinesis Data Streams の用語と概念

Kinesis Data Streamsに複数のシャードを作成した場合、配信先のシャードに偏りが発生しないようなパーティションキーを設定する必要があります。今回はハンズオンということで、後述の通り、シャード数を1に設定するため、統一的に「partitionkey」という文字列を指定します。

Kinesis Data Streamsにデータを配信するためには、kinesis:PutRecordの権限を付与する

Lambda関数用のIAMロールを定義します。

Resources:
  DataSourceLambdaRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Action: sts:AssumeRole
            Principal:
              Service:
                - lambda.amazonaws.com
      Policies:
        - PolicyName: PutRecordIntoKinesisDataStream
          PolicyDocument:
            Version: 2012-10-17
            Statement:
              - Effect: Allow
                Action:
                  - kinesis:PutRecord
                Resource:
                  - !Ref KinesisDataStreamArn
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
Code language: YAML (yaml)

ポイントはIAMロール内の、インラインポリシーで許可するアクションです。Kinesis Data Streamsにデータを配信するために必要な権限は「kinesis:PutRecord」です。同アクションをActionプロパティに指定します。また配信先として、Resourceプロパティに、後述のKinesis Data StreamsのストリームのARNを指定します。

Kinesis Data Streamsのスループットはシャード数で決まる

saa-04-001-kinesis.yamlでKinesis関係のリソースを定義します。まず後述のLambda関数で生成されたデータをキャプチャするKinesis Data Streamsを定義します。

Resources:
  KinesisDataStream:
    Type: AWS::Kinesis::Stream
    Properties:
      Name: KinesisDataStream
      ShardCount: 1
Code language: YAML (yaml)

Kinesis Data Streamsで重要なパラメータはシャードです。シャードはShardCountプロパティで設定します。シャードはKinesis Data Streamsのスループットを決定します。

シャードは、ストリーム内の一意に識別されたデータレコードのシーケンスです。ストリームは複数のシャードで構成され、各シャードが容量の 1 単位になります。各シャードは、読み取りに対して最大 5 トランザクション/秒をサポートし、最大合計データ読み取りレートは 2 MB /秒、最大合計データ読み取りレートは 1,000 レコード、最大合計データ書き込みレートは 1 秒あたり 1 MB（パーティションキーを含む）までサポートできます。ストリームのデータ容量は、ストリームに指定したシャードの数によって決まります。ストリームの総容量はシャードの容量の合計です。
Amazon Kinesis Data Streams の用語と概念

つまりシャード数を増やすことによって、Kinesis Data Streams全体のスループットを高めることができます。今回は検証ということで、先述の通り、シャード数は最小の「1」に設定します。

Kinesis Data Analytics用語

テンプレートファイルの具体的な記載内容を確認する前に、Kinesis Data Analyticsに関する用語を整理します。

Kinesis Data Analyticsで処理する対象のストリーミングデータのことを、ストリーミングソースと呼びます。今回の検証で言うと、Kinesis Data Streamsをストリーミングソースと呼びます。Kinesis Data Analyticsを起動すると、アプリケーション内ストームが生成されます。

アプリケーションを起動すると、Kinesis Data Analytics によって指定されたアプリケーション内ストリームが作成されます。この名前を使用して、アプリケーションコードでアプリケーション内ストリームにアクセスします。
アプリケーション入力の設定

つまりKinesis Data Analyticsで処理の対象とするデータは、Kinesis Data Streams等のストリーミングデータそのものではなく、Kinesis Data Analyticsが生成するアプリケーション内ストームであるということです。

1つのアプリケーション内ストームで処理できる速度には限度があります。

1 つのアプリケーション内ストリームに対して推奨される最大スループットは、アプリケーションのクエリの複雑さに応じて、2 ～ 20 MB/秒です。
Limits

それ以上の性能が必要の場合は、アプリケーション内ストームは複数作成することが可能です。

オプションで、ストリーミングリソースを複数のアプリケーション内ストリームにマッピングできます。 (中略) この場合、Amazon Kinesis Data Analytics では、次の名前で、アプリケーション内ストリームを指定された数だけ作成します。プレフィックス_001,プレフィックス_002, およびプレフィックス_003。
Limits

上記のように、アプリケーション内ストームは自動的に作成されますが、デフォルトで作成されたものに加え、手動で作成することも可能です。

必要に応じて、中間クエリ結果を保存するためにアプリケーション内ストリームをさらに作成することもできます。
アプリケーション内ストリームとポンプ

手動で作成したアプリケーション内ストームを使用するためには、デフォルトのアプリケーション内ストームから、データをポンプする必要があります。

ポンプとは、1 つのアプリケーション内ストリームから別のアプリケーション内ストリームにデータを挿入する、連続して実行される挿入クエリです。
アプリケーション内ストリームとポンプ

Kinesis Data AnalyticsはストリーミングデータをSQLでフィルタリングできる

用語がわかりましたので、実際にKinesis Data Analyticsを定義を確認します。

Resources:
  KinesisDataAnalytics:
    Type: AWS::KinesisAnalytics::Application
    Properties:
      ApplicationName: !Sub "${Prefix}-application"
      ApplicationDescription: sample application.
      ApplicationCode:
        !Sub |
          CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" (event_time VARCHAR(20), ticker VARCHAR(5), price REAL);
          CREATE OR REPLACE PUMP "STREAM_PUMP" AS INSERT INTO "DESTINATION_SQL_STREAM"
          SELECT STREAM event_time, ticker, price
          FROM "${NamePrefix}_001"
          WHERE ticker = 'AAPL';
      Inputs:
        - NamePrefix: !Ref NamePrefix
          InputSchema:
            RecordColumns:
              - Name: EVENT_TIME
                SqlType: VARCHAR(20)
                Mapping: $.EVENT_TIME
              - Name: TICKER
                SqlType: VARCHAR(5)
                Mapping: $.TICKER
              - Name: PRICE
                SqlType: REAL
                Mapping: $.PRICE
            RecordFormat:
              RecordFormatType: JSON
              MappingParameters:
                JSONMappingParameters:
                  RecordRowPath: $
          KinesisStreamsInput:
            ResourceARN: !GetAtt KinesisDataStream.Arn
            RoleARN: !GetAtt KinesisAnalyticsRole.Arn
Code language: YAML (yaml)

Inputsプロパティ内でストリーミングソースに関する設定を行います。

NamePrefixプロパティで、自動的に作成されるアプリケーション内ストリームのプレフィックスを指定します。先述の通り、ストリーム名には数字のサフィックスが付与されますので、今回作成されるストリーム名は「SAMPLE_APP_001」となります。

KinesisStreamsInputプロパティでストリーミングソースを指定します。ResourceARNプロパティに、定義済みのKinesis Data StreamsのストリームのARNを指定します。RoleARNプロパティに、同ストリームにアクセスするために必要な権限をまとめたIAMロールのARNを指定します。
RecordFormatプロパティは、ストリーミングソースから受け取るデータに関する設定を行います。RecordFormatプロパティでは受け取るデータの形式を指定します。今回の検証用のデータは、Lambda関数からJSON形式で配信されるため、「JSON」を指定します。MappingParametersプロパティでは、後述のパラメータマッピング用として、1つのデータを意味する変数を定義します。今回は、「$」で1データを示すことにします。

RecordColumnsプロパティで、ストリーミングソースから受け取るデータ内の要素と、アプリケーション内ストリーム内部のカラムのマッピングを定義します。今回は1データに3つの情報(日時、ティッカー、株価)を持っていますので、3つのプロパティでそれらを割り当てます。Nameプロパティはアプリケーション内ストリームに作成するカラムです。SqlTypeプロパティはカラムの型です。取れる型はData Typesに詳しいですが、今回は、文字列はVARCHAR、数値はREALを指定します。Mappingプロパティで、カラムが参照する元データを指定します。先述のマッピング用変数「$」を使用します。

ApplicationCodeプロパティで実行するSQLを定義します。1行目の「CREATE OR REPLACE STREAM」文で、新しいアプリケーション内ストリーム「DESTINATION_SQL_STREAM」を作成します。デフォルトのアプリケーション内ストリーム「SAMPLE_APP_001」と同様に、3カラムを定義します。2行目の「CREATE OR REPLACE PUMP」文で、ポンプを作成します。「STREAM_PUMP」という名前のポンプを作成します。ポンプでアプリケーション内ストリーム間でデータを連携します。今回は先ほど作成した「DESTINATION_SQL_STREAM」に連携するように指定します。3~5行目でポンプで連携するカラムおよび連携元を指定します。今回はFROM句で「SAMPLE_APP_001」を連携元のストリームを指定し、3カラムを「DESTINATION_SQL_STREAM」に渡すように指定します。加えてWHERE句でティッカー名が「AAPL」のデータに抽出するように指定します。