Amazon Kinesis Data StreamsにTwitter検索データを送信する

2021/08/09 2021/08/09

Kinesis Data Streamsの作成

ストリーム名とシャード数を決定するだけです。

シャードは
「各シャードは最大で 1 MiB/秒および 1000 レコード/秒を取得し、最大 2 MiB/秒を出力します。」
ということですので、1レコード1kbなさそうなレコードを最大100件送信するのでシャードは1つで足りそうです。

シャードエスティメーターで確認しても問題なさそうです。

作成されました。
シャードの数、タグ以外の設定は、サーバー側の暗号化、データ保持期間、シャードレベルメトリクスでした。
デフォルトの24時間、暗号化なし、シャードレベルの追加メトリクスなしにしました。

AWS CLI を使用した基本的な Kinesis Data Stream オペレーションの実行の手順でレコードの送信、受信をやってみました。

送信

$ aws kinesis put-record --stream-name Tweet --partition-key 123 --data testdata
{
    "ShardId": "shardId-000000000000",
    "SequenceNumber": "49620429506074472818275695973080444976867703729602691074"
}

$ aws kinesis put-record --stream-name Tweet --partition-key 123 --data testdata

{

"ShardId": "shardId-000000000000",

"SequenceNumber": "49620429506074472818275695973080444976867703729602691074"

}

受信

$ aws kinesis get-records --shard-iterator $(aws kinesis get-shard-iterator --shard-id shardId-000000000000 --shard-iterator-type TRIM_HORIZON --stream-name Tweet --query 'ShardIterator')
{
    "Records": [
        {
            "SequenceNumber": "49620429506074472818275695973080444976867703729602691074",
            "ApproximateArrivalTimestamp": "2021-07-24T09:18:19.003000+00:00",
            "Data": "testdata",
            "PartitionKey": "123"
        }
    ],
    "NextShardIterator": "AAAAAAAAAAEE7PmjeZZuA9VoWDyAORnsK9ZmBAlCWRcJOtVKQh7wvlX7AnNli52adZZeuy6pk09sDDCmwopXBCXkdobIPLj642Il3RgV25muolgT4dmP37ob7Qa0baH2EParQtmbEcdBmVZddONbqUXvfoVS13Y2o9btm3JoYb0Y5KL/paZl6nZAmm5ruDYz03mJnFX+iqeO8M4SML2ZNYUApx3uB/QjcGi3uJgEDjSNAnQyVYv55w==",
    "MillisBehindLatest": 0
}

$ aws kinesis get-records --shard-iterator $(aws kinesis get-shard-iterator --shard-id shardId-000000000000 --shard-iterator-type TRIM_HORIZON --stream-name Tweet --query 'ShardIterator')

{

"Records": [

{

"SequenceNumber": "49620429506074472818275695973080444976867703729602691074",

"ApproximateArrivalTimestamp": "2021-07-24T09:18:19.003000+00:00",

"Data": "testdata",

"PartitionKey": "123"

}

"NextShardIterator": "AAAAAAAAAAEE7PmjeZZuA9VoWDyAORnsK9ZmBAlCWRcJOtVKQh7wvlX7AnNli52adZZeuy6pk09sDDCmwopXBCXkdobIPLj642Il3RgV25muolgT4dmP37ob7Qa0baH2EParQtmbEcdBmVZddONbqUXvfoVS13Y2o9btm3JoYb0Y5KL/paZl6nZAmm5ruDYz03mJnFX+iqeO8M4SML2ZNYUApx3uB/QjcGi3uJgEDjSNAnQyVYv55w==",

"MillisBehindLatest": 0

}

shard-iteratorを入れ子コマンドで取得しました。

テストで作ったレコードが残っているとツイッターのデータが書き込めなかったので、一回ストリームを削除して作り直しました。

IAMロールの設定

AWS Lambdaから送信するので、専用のIAMロールを設定しました。

KinesisPutRecords

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "kinesis:PutRecords",
                "kinesis:PutRecord"
            ],
            "Resource": "arn:aws:kinesis:us-east-1:123456789012:stream/Tweet"
        },
        {
            "Effect": "Allow",
            "Action": "secretsmanager:GetSecretValue",
            "Resource": "arn:aws:secretsmanager:us-east-1:123456789012:secret:LiveDelivery/TwitterAPIKey-xxxxx"
        },
        {
            "Effect": "Allow",
            "Action": [
                "ssm:PutParameter",
                "ssm:GetParameter"
            ],
            "Resource": "arn:aws:ssm:us-east-1:123456789012:parameter/twitter-since-id"
        },
        {
            "Effect": "Allow",
            "Action": [
                "xray:PutTraceSegments",
                "xray:PutTelemetryRecords"
            ],
            "Resource": [
                "*"
            ]
        }
    ]
}

{

"Version": "2012-10-17",

"Statement": [

{

"Effect": "Allow",

"Action": [

"kinesis:PutRecords",

"kinesis:PutRecord"

"Resource": "arn:aws:kinesis:us-east-1:123456789012:stream/Tweet"

{

"Effect": "Allow",

"Action": "secretsmanager:GetSecretValue",

"Resource": "arn:aws:secretsmanager:us-east-1:123456789012:secret:LiveDelivery/TwitterAPIKey-xxxxx"

{

"Effect": "Allow",

"Action": [

"ssm:PutParameter",

"ssm:GetParameter"

"Resource": "arn:aws:ssm:us-east-1:123456789012:parameter/twitter-since-id"

{

"Effect": "Allow",

"Action": [

"xray:PutTraceSegments",

"xray:PutTelemetryRecords"

"Resource": [

"*"

]

}

]

}

kinesis:PutRecordsを使うことにしました。
必要ないですが、PutRecordも許可しています。
TwitterのキーをSecretsManagerから取得するポリシーとParameterStoreでsince_idを管理しているのでそのポリシーも設定しています。

他には、AWS管理ポリシーのAWSLambdaBasicExecutionRoleを設定しています。

Lambda関数

Githubで公開しています。
yamamanx/TweetSearchAnalyze

取得したTweetをput_recordでKinesis Data Streamingに送信しています。

def kinesis_put_records(tweets):
    try:
        records = []
        for tweet in tweets['statuses']:
            payload = {
                'id': tweet['id_str'],
                'text': tweet['text'],
                'source': tweet['source'],
                'user_name': tweet['user']['name'],
                'user_id': tweet['user']['id'],
                'user_screen_name': tweet['user']['screen_name'],
                'user_followers_count': tweet['user']['followers_count'],
                'geo': tweet['geo']
            }
            logger.debug(payload)
            record = {
                'Data': json.dumps(payload),
                'PartitionKey': tweet['id_str']
            }
            records.append(record)

        logger.info(len(records))
        if len(records) > 0:
            kinesis_client = boto3.client('kinesis')
            response = kinesis_client.put_records(
                Records=records,
                StreamName='Tweet'
            )
            logger.info(response)

    except Exception as e:
        raise e