How to read a CSV file from s3 and write the content in RDS database table using python lambda function?

Question

I have a CSV file Employee.csv in the S3 bucket with all info about employee: name, age, salary, designation. I have to write a python lambda function to read this file and write in RDS db such as it should create a table as Employee, with columns name, age, salary, designation and rows will have the data.

The Employee.csv is just for example, actually it can be any csv file with any number of columns in it.

rmaleki · Accepted Answer · 2020-01-28 09:56:55Z

2

from __future__ import print_function
import boto3
import logging
import os
import sys
import uuid
import pymysql
import csv
import rds_config


rds_host  = rds_config.rds_host
name = rds_config.db_username
password = rds_config.db_password
db_name = rds_config.db_name


logger = logging.getLogger()
logger.setLevel(logging.INFO)

try:
    conn = pymysql.connect(rds_host, user=name, passwd=password, db=db_name, connect_timeout=5)
except Exception as e:
    logger.error("ERROR: Unexpected error: Could not connect to MySql instance.")
    logger.error(e)
    sys.exit()

logger.info("SUCCESS: Connection to RDS mysql instance succeeded")

s3_client = boto3.client('s3')

def handler(event, context):

    bucket = event['Records'][0]['s3']['bucket']['name']
    key = event['Records'][0]['s3']['object']['key'] 
    download_path = '/tmp/{}{}'.format(uuid.uuid4(), key)

    s3_client.download_file(bucket, key,download_path)

    csv_data = csv.reader(file( download_path))

    with conn.cursor() as cur:
        for idx, row in enumerate(csv_data):

            logger.info(row)
            try:
                cur.execute('INSERT INTO target_table(name, age, salary, designation)' \
                                'VALUES("%s", "%s", "%s", "%s")'
                                , row)
            except Exception as e:
                logger.error(e)

            if idx % 100 == 0:
                conn.commit()

        conn.commit()

    return 'File loaded into RDS:' + str(download_path)

answered Jan 28, 2020 at 9:56

rmaleki

7822 gold badges12 silver badges34 bronze badges

Sign up to request clarification or add additional context in comments.

7 Comments

Kapil Shukla Over a year ago

is it necessary to download the file? s3_client.download_file(bucket, key,download_path)

rmaleki Over a year ago

no, it is not necessary. you can open CSV from local.

Kapil Shukla Over a year ago

I have another doubt, is there a create table command needed before insert table?

rmaleki Over a year ago

yes, there is. you should create a table in your SQL database. for example, write this command in postgress bash: CREATE TABLE table_name ( column_name TYPE column_constraint, table_constraint table_constraint ) INHERITS existing_table_name;

Kapil Shukla Over a year ago

getting this error : not enough arguments for format string. for the INSERT statement

|

Kapil Shukla · Accepted Answer · 2020-02-11 12:52:38Z

Here is the code which is working for me now:

s3 = boto3.resource('s3')
    file_object=event['Records'][0]
    key=str(file_object['s3']['object']['key'])
    obj = s3.Object(bucket, key)

    content_lines=obj.get()['Body'].read().decode('utf-8').splitlines(True)

    tableName= key.strip('folder/').strip('.csv')

    with conn.cursor() as cur:
        try:
            cur.execute('TRUNCATE TABLE '+tableName)
        except Exception as e:
            print("ERROR: Unexpected error:Table does not exit.")
            sys.exit()        
        header=True
        for row in csv.reader(content_lines):
            if(header):
                numberOfColumns=len(row)
                columnNames= str(row).replace('[','').replace(']','').replace("'",'')
                print("columnNames:"+columnNames)
                values='%s'
                numberOfValues=len(values)
                numberOfValues=1
                while numberOfValues< numberOfColumns:    
                    values=values+",%s"
                    numberOfValues+=1
                print("INSERT into "+tableName+"("+columnNames+") VALUES("+values+")")
                header=False
            else:
                try:
                    cur.execute('INSERT into '+tableName+'('+columnNames+') VALUES('+values+')', row)
                except Exception as e:
                    raise e
        conn.commit()

Collectives™ on Stack Overflow

How to read a CSV file from s3 and write the content in RDS database table using python lambda function?

2 Answers 2

7 Comments

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

7 Comments

Comments

Your Answer

Sign up or log in

Post as a guest

Related