AWS Kinesis Analyticsでレコードの事前処理(Lambda Python 3.7)

## はじめに
Amazon Kinesis Data Analyticsでは、**Record pre-processing**を**Enabled**にすることで、**Lambda**でレコードの**事前処理**ができます。

- Lambda 関数を使用したデータの事前処理  
https://docs.aws.amazon.com/ja_jp/kinesisanalytics/latest/dev/lambda-preprocessing.html

Lambdaのサンプルコードはドキュメントによるとblueprintでテンプレートが用意されているはずなのですが、利用できない状態になっています。
（コンソールからリンクをクリックしても何も出てこない。。）

- 事前処理のための Lambda 関数の作成  
https://docs.aws.amazon.com/ja_jp/kinesisanalytics/latest/dev/lambda-preprocessing-functions.html

今回、Lambdaの事前処理について**Twitterのツイートの頻出単語ウインドウ集計**をテーマにしてサンプルを作成してみたのでまとめておきます。

## 構成
以下の順番で処理をしていきます。

1. LogstashでTwitter Streaming APIからツイートを取得。
2. LogstashからKinesis Data Streamsにツイートを転送。
3. Kinesis Data AnalyticsのData SourceをData Streamsとし、LambdaでRecord pre-processingする
4. Lambdaはツイートを単語間のスペースで分割してData Analyticsに返す
5. Data Analyticsは単語の出現数をウインドウ集計

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/427238/7eb3bd0b-ef35-fd97-cfde-a88b919481c5.png)

## Logstash

Logstashのconfファイルのサンプルです。Twitter APIからツイートを取得し、必要なフィールドのみをKinesis Data Streamsにアウトプットします。

```conf
input {
  twitter {
    consumer_key => ""
    consumer_secret => ""
    oauth_token => ""
    oauth_token_secret => ""
    languages => ["en"]
    full_tweet => true
    keywords => ["trump"]
    ignore_retweets => true
    codec => "json"
  }
}
filter {
  # "text"と"@timestamp"フィールドのみを残す
  ruby {
    code => "
      wanted_fields = ['text', '@timestamp']
      event.to_hash.keys.each { |k|
        event.remove(k) unless wanted_fields.include? k
      }
    "
  }
}
output {
   kinesis {
     stream_name => ""
     region => "ap-northeast-1"
     access_key => ""
     secret_key => ""
   }
}
```

## Record pre-processing

**Record pre-processing**を**Enabled**にして、対象のLambdaを設定します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/427238/87801fa1-e8d9-3765-7cd1-240ede5a2332.png)

## Lambda

Kinesis Streamsからレコードを受け取り、受け取ったレコードのTextを単語間のスペースで分割していきます。

ポイントは以下です。

#### dataフィールドはbase64エンコードが必要
受け取る時はデコード、アウトプットはエンコードしなければなりません。

- レコードレスポンスモデル  
https://docs.aws.amazon.com/ja_jp/kinesisanalytics/latest/dev/lambda-preprocessing.html#lambda-preprocessing-data-model

また、Kinesisをデータソースとした場合に限った話ではありませんが、Lambdaハンドラーのeventは辞書型に変換が必要です。(以下の記事参照)

- 【Tips】Lambda(Python)でハンドラーのeventを全て辞書型だと思っていたらハマったこと  
https://dev.classmethod.jp/cloud/aws/lambda-python-tips-all-events-are-not-dict/

```python:lambda_function.py
import json
import base64
import logging
import re
logger = logging.getLogger()
logger.setLevel(logging.INFO)

def lambda_handler(event, context):
    response = []
    for record in event['records']:
        wordlist = []

        try:
            # Base64で送られてきたレコードをデコードする
            data = base64.b64decode(record['data']).decode(encoding='utf-8', errors='replace')
            # dataを辞書型に変換
            data_json = json.loads(data)

            logger.info(data_json)

            # ツイートの分割
            splited_words = re.split(" +", data_json['text'])
            for word in splited_words:
                wordlist.append({"word" : word})

            logger.info(wordlist)

            # Base64にエンコード
            wordlist_json = json.dumps(wordlist)
            wordlist_byte = wordlist_json.encode('utf-8')
            wordlist_base64 = base64.b64encode(wordlist_byte)
            wordlist_base64_str = wordlist_base64.decode()

            # 戻り値のdataの生成
            response.append(
                {
                    "recordId" : record['recordId'],
                    "result" : "Ok",
                    "data": wordlist_base64_str
                }
            ) 
        # 処理に失敗した場合の戻り値の生成
        except:
            response.append(
                {
                    "recordId" : record['recordId'],
                    "result" : "Dropped"
                }
            )

    return_dict = {
        "records": response
    }

    logger.info(return_dict)
    # 戻り値は辞書型（JSONではない点に注意！）
    return return_dict
```

## Kinesis Analytics
### SQL

"word"フィールドを10秒毎に集計するSQLです。

```

CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" (word VARCHAR(32), word_count INTEGER);

CREATE OR REPLACE  PUMP "STREAM_PUMP" AS INSERT INTO "DESTINATION_SQL_STREAM"

SELECT STREAM "word", COUNT(*) AS "word_count"
FROM "SOURCE_SQL_STREAM_001"

GROUP BY "word", FLOOR(("SOURCE_SQL_STREAM_001".ROWTIME - TIMESTAMP '1970-01-01 00:00:00') SECOND / 10 TO SECOND);

```

### 集計結果
集計結果が確認できます。
"WORD_COUNT"列に集計した数が表示されています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/427238/80361d99-d1ff-f9fc-6652-6c71b852061d.png)


## 最後に

Lambdaの事前処理を紹介しました。
Base64の処理のところに手こずるかもしれませんが、事前処理がLambdaで手軽にできるのは便利かと思います。

誰かのお役に立てれば幸いです！