如何访问来自 AWS Kinesis Data Stream 事件的数据？答案

【问题标题】：How do I access the data from an AWS Kinesis Data Stream event?如何访问来自 AWS Kinesis Data Stream 事件的数据？
【发布时间】：2020-01-02 03:27:02
【问题描述】：

我正在开发一个使用 AWS Kinesis Data Stream 的 Python lambda。但我很难理解运动记录事件的形状。例如：

{
    "Records": [
        {
            "kinesis": {
                "kinesisSchemaVersion": "1.0",
                "partitionKey": "1",
                "sequenceNumber": "49590338271490256608559692538361571095921575989136588898",
                "data": "SGVsbG8sIHRoaXMgaXMgYSB0ZXN0Lg==",
                "approximateArrivalTimestamp": 1545084650.987
            },
            "eventSource": "aws:kinesis",
            "eventVersion": "1.0",
            "eventID": "shardId-000000000006:49590338271490256608559692538361571095921575989136588898",
            "eventName": "aws:kinesis:record",
            "invokeIdentityArn": "arn:aws:iam::123456789012:role/lambda-role",
            "awsRegion": "us-east-2",
            "eventSourceARN": "arn:aws:kinesis:us-east-2:123456789012:stream/lambda-stream"
        },
        {
            "kinesis": {
                "kinesisSchemaVersion": "1.0",
                "partitionKey": "1",
                "sequenceNumber": "49590338271490256608559692540925702759324208523137515618",
                "data": "VGhpcyBpcyBvbmx5IGEgdGVzdC4=",
                "approximateArrivalTimestamp": 1545084711.166
            },
            "eventSource": "aws:kinesis",
            "eventVersion": "1.0",
            "eventID": "shardId-000000000006:49590338271490256608559692540925702759324208523137515618",
            "eventName": "aws:kinesis:record",
            "invokeIdentityArn": "arn:aws:iam::123456789012:role/lambda-role",
            "awsRegion": "us-east-2",
            "eventSourceARN": "arn:aws:kinesis:us-east-2:123456789012:stream/lambda-stream"
        }
    ]
}

来源：Using AWS Lambda with Amazon Kinesis

我最初放在此对象中表示的运动流中的数据在哪里？以及如何访问这些数据？

【问题讨论】：

标签： python python-3.x amazon-web-services aws-lambda amazon-kinesis

【解决方案1】：

您放入流中的数据在每条记录的 kinesis.data 键上表示为 Base64 编码字符串。例如（截断）：

{
    "Records": [
        {
            "kinesis": {
                ...
                "data": "SGVsbG8sIHRoaXMgaXMgYSB0ZXN0Lg==",
                ...
            },
            ...
        },
        {
            "kinesis": {
                ...
                "data": "VGhpcyBpcyBvbmx5IGEgdGVzdC4=",
                ...
            },
            ...
        }
    ]
}

要访问数据，请遍历每个 Records 对象，然后 Base64 解码 kinesis.data 值。

import base64


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

    print(decoded_data)
    # Record 1: Hello, this is a test.
    # Record 2: This is only a test.

注意：此示例假定发送到 kinesis 流的数据在 kinesis b64 编码之前最初是 utf-8 编码的。

【讨论】：