使用AWS Lambda将RDS MySQL写入Kinesis时遇datetime序列化错误

问题核心：从MySQL查询返回的rows中包含datetime类型对象，Python标准JSON库默认无法序列化这类非原生数据类型，导致json.dumps()调用失败。

解决方案1：自定义JSON序列化逻辑 #

在json.dumps中通过default参数指定datetime对象的转换规则，比如转成ISO标准格式字符串：

修改Kinesis写入段代码：

import json
from datetime import datetime

# ... 其他原有代码保持不变 ...

response = kinesis.put_record(
    StreamName="test",
    Data=json.dumps(rows, default=lambda x: x.isoformat() if isinstance(x, datetime) else x),
    PartitionKey="1"
)

如果需要更严谨的类型判断，可以单独定义序列化函数：

def serialize_datetime(obj):
    if isinstance(obj, datetime):
        return obj.isoformat()
    raise TypeError("Unsupported data type")

response = kinesis.put_record(
    StreamName="test",
    Data=json.dumps(rows, default=serialize_datetime),
    PartitionKey="1"
)

解决方案2：查询时直接转换datetime为字符串 #

在MySQL查询语句中，用DATE_FORMAT函数将datetime字段直接转为字符串，避免后续Python端的类型处理：

# 替换为你实际的datetime字段名，比如create_time
cursor.execute('SELECT id, name, DATE_FORMAT(create_time, "%Y-%m-%d %H:%i:%s") FROM table LIMIT 10')

解决方案3：使用字典游标简化字段处理 #

将游标改为字典游标，返回的每行数据是键值对字典，方便精准定位datetime字段并转换：

cursor = connection.cursor(pymysql.cursors.DictCursor)
cursor.execute('SELECT * FROM table LIMIT 10')
rows = cursor.fetchall()

# 遍历转换datetime字段（替换为你表中的实际datetime字段名）
for row in rows:
    if 'create_time' in row and isinstance(row['create_time'], datetime):
        row['create_time'] = row['create_time'].isoformat()

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

额外提示：代码中的connection.commit()可以删除，因为你仅执行了SELECT查询，无数据修改操作，不需要提交事务。

内容的提问来源于stack exchange，提问作者talha babar