如何使用 Lambda 将 CSV 文件数据从 S3 加载到 MySQL RDS？

Question

enter image description here enter image description here 比方说 数据库名称 = EmployeeDB 在这个数据库中，我们有 5 个表，它们是 table1、table2、table3、table4 和 table5 我们还有 5 个 CSV 文件，它们是草图、配置文件、阅读、健康、错误

场景： 每当在 S3 存储桶中上传 CSV 文件时，它应该触发该 CSV 数据并将其加载到特定表中。 （例如：当 sketching 上传时，它应该转到 table1 表格）

为了实现这一点，我尝试了 Lambda 函数，这是我使用的代码。

import json
import boto3
import csv
import mysql.connector
from mysql.connector import Error
from mysql.connector import errorcode
s3_client = boto3.client('s3')

# Read CSV file content from S3 bucket
def lambda_handler(event, context):
    # TODO implement
    # print(event)
    bucket = event['Records'][0]['s3']['bucket']['name']
    csv_file = event['Records'][0]['s3']['object']['key']
    csv_file_obj = s3_client.get_object(Bucket=bucket, Key=csv_file)
    lines = csv_file_obj['Body'].read().decode('utf-8').split()
    
    results = []
    for row in csv.DictReader(lines):
        results.append(row.values())
    print(results)
    
    connection = mysql.connector.connect(host='xxxxxxxxxxxxxxx.ap-south-1.rds.amazonaws.com',database='employeedb',user='xxxxxx',password='xxxxxx')
    
    tables_dict = {
        'sketching': 'INSERT INTO table1 (empid, empname, empaddress) VALUES (%s, %s, %s)',
        'profile': 'INSERT INTO table2 (empid, empname, empaddress) VALUES (%s, %s, %s)',
        'reading': 'INSERT INTO table3 (empid, empname, empaddress) VALUES (%s, %s, %s)',
        'health': 'INSERT INTO table4 (empid, empname, empaddress) VALUES (%s, %s, %s)',
        'error': 'INSERT INTO table5 (empid, empname, empaddress) VALUES (%s, %s, %s)'
    }
    if csv_file in tables_dict:
        mysql_empsql_insert_query = tables_dict[csv_file]
        cursor = connection.cursor()
        cursor.executemany(mysql_empsql_insert_query,results)
        connection.commit()
        print(cursor.rowcount, f"Record inserted successfully from {csv_file} file")
    return {
        'statusCode': 200,
        'body': json.dumps('Hello from Lambda!')
    }

此代码对我不起作用，它在 cloudwatch 中触发，但在我上传草图时它没有将数据加载到 table1 或任何表中。

谁能帮我修改我的场景的代码？

Answer 1

您可以使用您的文件和查询创建字典。对于每个文件/表，都需要不同的插入查询。

tables_dict = {
  'sketching.csv': 'INSERT INTO table1.sketching (empid, empname, empaddress) VALUES (%s, %s, %s)',
  'profile.csv': '',
  'reading.csv': '',
  'health.csv': '',
  'error.csv': ''
}

if csv_file in tables_dict:
  mysql_empsql_insert_query = tables_dict[csv_file]
  cursor = connection.cursor()
  cursor.executemany(mysql_empsql_insert_query,results)
  connection.commit()
  print(cursor.rowcount, f"Record inserted successfully from {csv_file} file")