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

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)

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"
    )

Python相关问答推荐

如何根据日期和时间将状态更新为已过期或活动？

将DF中的名称与另一DF拆分并匹配并返回匹配的公司

pandas DataFrame GroupBy.diff函数的意外输出

如何让剧作家等待Python中出现特定cookie(然后返回它)？

如何使用LangChain和AzureOpenAI在Python中解决AttribeHelp和BadPressMessage错误？

pandas滚动和窗口中有效观察的最大数量

如何合并两个列表，并获得每个索引值最高的列表名称？

在www.example.com中使用`package_data`包含不包含__init__. py的非Python文件

使用BeautifulSoup抓取所有链接

Matplotlib中的字体权重

(Python/Pandas)基于列中非缺失值的子集DataFrame

交替字符串位置的正则表达式

在Python中从嵌套的for循环中获取插值

Polars Group by描述扩展

如何求相邻对序列中元素 Select 的最小代价

删除特定列后的所有列

PySpark：如何最有效地读取不同列位置的多个CSV文件

如何在Python中自动创建数字文件夹和正在进行的文件夹？

如果不使用. to_list()[0]，我如何从一个pandas DataFrame中获取一个值？

如何在Python中创建仅包含完整天数的月份的列表