AWS Lambda无法访问MySQL RDS问题排查求助

问题背景 #

搭建的架构逻辑为：IRS数据文件上传至S3存储桶后触发Lambda运行，Lambda读取S3文件并写入MySQL RDS。本地环境可正常连接RDS，但Lambda始终触发超时错误，日志显示S3请求发起后超时。

错误日志 #

"errorMessage": "2022-11-15T22:22:51.919Z 9f20c035-5a47-4c6f-be9f-407b4a43aee6 Task timed out after 60.06 seconds"

关键日志片段：

[DEBUG] 2022-11-15T22:21:53.459Z    9f20c035-5a47-4c6f-be9f-407b4a43aee6    Starting new HTTPS connection (1): irs-data.s3.amazonaws.com:443
2022-11-15T22:22:51.919Z 9f20c035-5a47-4c6f-be9f-407b4a43aee6 Task timed out after 60.06 seconds

Lambda代码 #

import pandas as pd
import boto3
import os
from dotenv import load_dotenv
import logging
import sys
import time
import datetime as dt
import io
import pymysql

####### LOADING ENVIRONMENT VARIABLES #######
load_dotenv()
logger = logging.getLogger()
logger.setLevel(logging.DEBUG)

BUCKET = os.getenv('BUCKET')
BUCKET_PREFIX = os.getenv('BUCKET_PREFIX')

# Credentials to database connection
hostname= os.getenv('HOSTNAME')
dbname= os.getenv('DATABASE')
uname= os.getenv('USERNAME')
pwd= os.getenv('PASSWORD')

def lambda_handler(event, context):
    try:
        logger.info("TEST")
        logger.info(BUCKET)

        s3 = boto3.resource('s3')

        # assigning the bucket:
        my_bucket = s3.Bucket(BUCKET)

        data_list = []

        for my_bucket_object in my_bucket.objects.filter(Prefix=BUCKET_PREFIX):
            if my_bucket_object.key.endswith(".csv"):
                key=my_bucket_object.key
                body=my_bucket_object.get()['Body'].read()
                temp_data = pd.read_csv(io.BytesIO(body))
                data_list.append(temp_data)
        
        # concatenating all the files together:
        df = pd.concat(data_list)

        # Connect to MySQL Database
        connection = pymysql.connect(host=hostname,user=uname,password=pwd,database=dbname)

        cursor = connection.cursor()

        # Truncate the table everytime before an ETL:
        sql_trunc = "TRUNCATE TABLE `irs990`"
        cursor.execute(sql_trunc)

        # commit the results
        connection.commit()

        # creating columns from the dataframe:
        cols = "`,`".join([str(i) for i in df.columns.tolist()])

        # adding dataframe to mysql RDS
        for i,row in df.iterrows():
            sql = "INSERT INTO `irs990` (`" +cols + "`) VALUES (" + "%s,"*(len(row)-1) + "%s)"
            cursor.execute(sql, tuple(row))
            connection.commit()

        # checking if data was successfully written:
        sql = "SELECT * FROM `irs990`"
        cursor.execute(sql)
        result = cursor.fetchall()
        for i in result:
            print(i)

        # closing MySQL connection:
        connection.close()

    except Exception as e:
        logging.error(e)

排查步骤与解决方案 #

1. 修复Lambda访问S3的网络问题（核心超时原因）

Lambda配置VPC后默认失去公网访问能力，无法直接访问公网S3桶，这是日志中S3请求超时的直接原因：

• 方案A：配置S3 VPC端点
  在VPC中创建Gateway类型的S3端点，关联Lambda所在子网和路由表，让Lambda通过VPC内网直接访问S3，无需依赖公网。
• 方案B：配置NAT网关（适用于需要访问公网的场景）
  若Lambda使用私有子网，需为子网配置NAT网关，并确保路由表包含指向NAT网关的0.0.0.0/0路由；若使用公有子网，需确保子网路由表有互联网网关路由，且Lambda的ENI分配了公网IP（Lambda在公有子网默认不分配公网IP，需手动配置）。

2. 修正RDS安全组授权逻辑

不要使用Lambda动态出站IP授权（IP会随时变化），改用安全组ID授权：

• 在RDS安全组的入站规则中，添加Lambda所在安全组的ID，允许TCP 3306端口流量。
• 确认Lambda安全组的出站规则，允许TCP 3306（MySQL）和TCP 443（S3 HTTPS）的流量。

3. 代码优化与调试

• 移除load_dotenv()依赖：Lambda的环境变量通过控制台配置，直接使用os.environ.get("KEY")获取，避免本地.env文件在Lambda环境中无法加载导致参数错误。
• 添加数据库连接超时参数：缩短连接超时时间，快速定位连通性问题：

  connection = pymysql.connect(
      host=hostname,
      user=uname,
      password=pwd,
      database=dbname,
      connect_timeout=10  # 快速反馈连接结果
  )
  

• 拆分测试逻辑：先编写极简函数测试RDS连通性，排除S3干扰：

  import pymysql
  import os
  
  def lambda_handler(event, context):
      try:
          connection = pymysql.connect(
              host=os.environ['HOSTNAME'],
              user=os.environ['USERNAME'],
              password=os.environ['PASSWORD'],
              database=os.environ['DATABASE'],
              connect_timeout=5
          )
          print("RDS连接成功")
          connection.close()
          return {"status": "success"}
      except Exception as e:
          print(f"连接失败: {str(e)}")
          return {"status": "failed", "error": str(e)}
  

• 批量插入优化：替换逐行插入逻辑，减少数据库交互次数，提升性能：

  # 替换原逐行插入代码
  values = [tuple(row) for _, row in df.iterrows()]
  sql = f"INSERT INTO `irs990` (`{cols}`) VALUES ({','.join(['%s']*len(df.columns))})"
  cursor.executemany(sql, values)
  connection.commit()
  

4. 调整Lambda资源配置

• 提升内存至256MB或更高：Pandas处理数据需要较多内存，内存不足会导致性能下降；Lambda的CPU资源随内存比例分配，提升内存可显著加快数据处理速度。
• 临时调整超时时间至90秒：排查阶段延长超时时间，便于捕获更详细的错误信息，问题解决后再调回合理值。

内容的提问来源于stack exchange，提问作者Bilal Jamil