AWS Lambda中pandas.to_sql创建MySQL表但未插入数据的问题

排查与解决步骤 #

1. 确认DataFrame是否包含有效数据 #

预处理后的DataFrame可能为空，导致无数据插入。在to_sql前添加日志输出数据行数，验证数据是否存在：

import logging
logging.basicConfig(level=logging.INFO)

def connect_mysql(credentials: dict, dataframe: pd.DataFrame, table_name: str):
    # 新增日志：输出DataFrame行数
    logging.info(f"待插入数据行数: {dataframe.shape[0]}")
    if dataframe.empty:
        return "DataFrame为空，无数据可插入"
    
    engine = f'mysql+pymysql://{credentials["user"]}:{credentials["passwd"]}@{credentials["host"]}:{credentials["port"]}/{credentials["db"]}'
    engine_corp = create_engine(engine, paramstyle="format")
    connection = engine_corp.connect()
    dataframe.to_sql(name=table_name, con=connection, if_exists='replace', index=False)
    return f'Inserted to {table_name}'

查看CloudWatch日志中的行数输出，确认是否有数据。

2. 显式提交事务并管理连接 #

Lambda环境中，SQLAlchemy连接的事务可能未自动提交，导致数据回滚。修改代码显式提交事务，并使用with语句自动管理连接生命周期：

def connect_mysql(credentials: dict, dataframe: pd.DataFrame, table_name: str):
    engine = f'mysql+pymysql://{credentials["user"]}:{credentials["passwd"]}@{credentials["host"]}:{credentials["port"]}/{credentials["db"]}'
    engine_corp = create_engine(engine, paramstyle="format")
    
    # 使用with语句自动处理连接关闭
    with engine_corp.connect() as connection:
        dataframe.to_sql(name=table_name, con=connection, if_exists='replace', index=False)
        # 显式提交事务
        connection.commit()
    
    return f'Inserted to {table_name}'

3. 检查数据类型兼容性 #

DataFrame列类型与MySQL表列类型不匹配时，可能静默失败。可以显式指定dtype参数映射类型，避免兼容性问题：

from sqlalchemy.types import String, DateTime, Integer

def connect_mysql(credentials: dict, dataframe: pd.DataFrame, table_name: str):
    engine = f'mysql+pymysql://{credentials["user"]}:{credentials["passwd"]}@{credentials["host"]}:{credentials["port"]}/{credentials["db"]}'
    engine_corp = create_engine(engine, paramstyle="format")
    
    # 根据实际列定义类型映射
    dtype = {
        "string_col": String(255),
        "date_col": DateTime(),
        "int_col": Integer()
    }
    
    with engine_corp.connect() as connection:
        dataframe.to_sql(
            name=table_name, 
            con=connection, 
            if_exists='replace', 
            index=False,
            dtype=dtype
        )
        connection.commit()
    
    return f'Inserted to {table_name}'

4. 增加详细执行日志 #

在关键节点添加日志，确认to_sql执行流程是否完整：

def connect_mysql(credentials: dict, dataframe: pd.DataFrame, table_name: str):
    logging.info(f"开始处理表{table_name}的数据插入")
    logging.info(f"DataFrame列信息: {list(dataframe.columns)}")
    
    engine = f'mysql+pymysql://{credentials["user"]}:{credentials["passwd"]}@{credentials["host"]}:{credentials["port"]}/{credentials["db"]}'
    engine_corp = create_engine(engine, paramstyle="format")
    
    with engine_corp.connect() as connection:
        logging.info("已建立MySQL连接")
        dataframe.to_sql(name=table_name, con=connection, if_exists='replace', index=False)
        connection.commit()
        logging.info("事务已提交")
    
    logging.info(f"完成表{table_name}的数据插入")
    return f'Inserted to {table_name}'

通过CloudWatch日志确认每一步是否执行成功。

5. 验证MySQL用户权限 #

虽然能创建表，但需确保用户拥有INSERT权限。登录MySQL执行以下命令验证：

SHOW GRANTS FOR 'your_user'@'your_host';

如果缺少INSERT权限，执行授权：

GRANT INSERT ON your_database.* TO 'your_user'@'your_host';
FLUSH PRIVILEGES;

内容的提问来源于stack exchange，提问作者Yusif Abasov