RDS MySQL 到 Kinesis 数据流的管道，使用 AWS Lambda。

Question

我正在尝试使用 put_record boto3 API 从 RDS MySQL 实例中提取数据并加载到 kinesis 数据流中。使用 pymysql 进行连接是可行的，并且我能够打印表格，但是无法将数据写入 kinesis 数据流。我遇到了这个错误："Object of type datetime is not JSON serializable"。

def lambda_handler(event, context):
    
    connection = pymysql.connect(
    host = endpoint, 
    user = username,
    password = passwrd,
    database = database_name)
    
    cursor = connection.cursor()
    cursor.execute('SELECT * FROM table LIMIT 10')

    rows = cursor.fetchall()

    for row in rows:
        print("{0} {1} {2}".format(row[0], row[1], row[2]))

    kinesis = boto3.client('kinesis')

    response = kinesis.put_record(
        StreamName="test",
        Data=json.dumps(rows),
        PartitionKey="1"
    )
    
    connection.commit()
    
lambda_handler(None,None)

我尝试打印表格，它成功了。唯一的问题是将记录放入Kinesis数据流中。

- talha babar

Answer 1

当将一个具有datetime类型属性的对象传递给json.dumps方法时，会抛出此错误。有几种解决方法，但最快的方法是使用json.dumps的keyword参数将所有datetime属性转换为字符串。

import json
from datetime import datetime

# The default keyword argument can 
# be set to a function that gets called 
# for objects that can't otherwise be serialized.
    response = kinesis.put_record(
        StreamName="test",
        Data=json.dumps(rows, default=str),
        PartitionKey="1"
    )

- Allan Chua