在 Lambda 上使用 python 从 S3 逐行处理 CSV

Question

我正在尝试使用 AWS Lambda (Python) 处理 S3 存储桶上的 .csv (30MB) 文件。我在本地编写了 python 代码来处理文件，现在尝试使用 Lambda 执行。很难逐行读取文件。

请告诉我如何使用 boto3 或 s3 方法逐行遍历文件。请尽快帮助我。谢谢

在 Lambda 中：

s3 = boto3.client("s3")
        file_obj = event["Records"][0]
        filename=str(file_obj['s3']['object']['key'])
        #print('file name is :', filename)
        fileObj = s3.get_object(Bucket=<mybucket>, Key=filename)
        file_content = fileObj["Body"].read().decode('utf-8')

我的原始代码：

import csv
import pandas as pd
import datetime
#from datetime import datetime,timedelta
import numpy as np
with open ('sample.csv', 'r') as file_name:

    csv_reader = csv.reader(file_name, delimiter=',')
    Time = []
    Latitude=[]
    Longitude= []
    Org_Units=[]
    Org_Unit_Type =[]
    Variable_Name=[]
    #New columns
    Year=[]
    Month= []
    Day =[]
    Celsius=[]
    Far=[]
    Conv_Units=[]
    Conv_Unit_Type=[]
    header = ['Time','Latitude', 'Longitude','Org_Units','Org_Unit_Type','Conv_Units','Conv_Unit_Type','Variable_Name']
    out_filename = 'Write' + datetime.datetime.now().strftime("%Y%m%d-%H%M%S") #need to rename based on the org file name

    with open(out_filename +'.csv', 'w') as csvFile:
        outputwriter = csv.writer(csvFile, delimiter=',')
        outputwriter.writerow(header)
        next(csv_reader, None)  # avoid hearder

        for row in csv_reader:
           # print(row)
            Time = row[0]
            Org_Lat=row[1]
            Org_Long=row[2]
            Org_Units=row[3]
            Org_Unit_Type =row[4]
            Variable_Name=row[5]
            # print(Time,Org_Lat,Org_Long,Org_Units,Org_Unit_Type,Variable_Name)

            if Org_Unit_Type == 'm s-1':
                Conv_Units =round(float(Org_Units) * 1.151,2)
                Conv_Unit_Type = 'miles'
            if Org_Unit_Type == 'm':
                Conv_Units =round(float(Org_Units) / 1609.344,2)
                 # print (Org_Units,Conv_Units)
                Conv_Unit_Type = 'miles'
            if Org_Unit_Type == 'Pa':
                Conv_Units =round(float(Org_Units) / 6894.757,2)
                Conv_Unit_Type = 'Psi'
                #print(type(Time))
            date_time_obj = datetime.datetime.strptime(Time, '%m-%d-%Y, %H:%M')
             #  Year = time.strptime(date_time_obj, "%B")
            #print(date_time_obj)
            f_row =[Time,Latitude,Longitude,Org_Units,Org_Unit_Type,Conv_Units,Conv_Unit_Type,Variable_Name]
            outputwriter.writerow(f_row)
csvFile.close()
print("done")

Answer 1

我认为这应该有效，您唯一需要检查的是您的 lambda 需要一个角色，该角色具有对 s3 存储桶具有读取访问权限的策略。 最初为了测试，我会在 s3 上完全访问 lambda AmazonS3FullAccess

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": "s3:*",
            "Resource": "*"
        }
    ]
}

python 代码

s3 = boto3.client('s3')
def lambda_handler(event, context):
    # Get the object from the event and show its content type
    bucket = event['Records'][0]['s3']['bucket']['name']
    key = event['Records'][0]['s3']['object']['key'].encode('utf8')
    obj = s3.get_object(Bucket=bucket, Key=key)
    rows    = obj['Body'].read().split('\n')
    print("rows" + rows)

Answer 2

与使用.read() 将对象作为流读取相比，您可能会发现将对象下载到本地存储更容易：

s3_client = boto3.client('s3', region='ap-southeast-2')
s3_client.download_file(bucket, key, '/tmp/local_file.csv')

然后您可以使用您的原始程序来处理该文件。

完成后，请务必删除临时文件，因为 AWS Lambda 容器可能会被重复使用，并且只有 500MB 可用磁盘空间。