如何使用Lambda与Python在AWS S3同文件夹读写文件且不触发循环

问题根因 #

你当前为Lambda配置的触发规则是监听FolderA/路径下的所有对象创建事件，当你处理完文件后覆盖原路径的对象时，本次覆盖操作会再次触发对象创建事件，导致Lambda被循环调用。

解决方案（支持保存回原文件夹原文件名） #

在代码中新增自定义元数据标记逻辑：每次Lambda被触发时先读取目标S3对象的元数据，若对象已经被标记为已处理，则直接终止执行；处理完成后上传文件时，给对象添加已处理标记，避免后续被重复触发。

修改后的完整代码如下： #

import json
import urllib.parse
import boto3
import os
import csv
import io
print('Loading function')
s3 = boto3.client('s3')

def lambda_handler(event, context):
    # 从事件中获取桶名和文件路径
    bucket = event['Records'][0]['s3']['bucket']['name']
    key = urllib.parse.unquote_plus(event['Records'][0]['s3']['object']['key'], encoding='utf-8')
    try:
        print("file name " + key)
        # 先读取对象元数据，判断是否已经处理过
        obj_head = s3.head_object(Bucket=bucket, Key=key)
        if obj_head['Metadata'].get('processed', 'false') == 'true':
            print(f"文件{key}已处理过，直接跳过")
            return True
        
        file_key = key
        csvfile = s3.get_object(Bucket=bucket, Key=file_key)
        csvcontent = csvfile["Body"].read().decode("utf-8")
        file = csvcontent.split("\n")
        csv_reader = csv.reader(file)
        line_count = 0
        colindex = ''
        content = []
        s33 = boto3.resource('s3')
        copy_source = {
              'Bucket': bucket,
              'Key': file_key
            }
        new_bucket = s33.Bucket(bucket)
        src_folder = "FolderA/"
        new_filekey = file_key.replace(src_folder,"")
        new_bucket.copy(copy_source, 'BKP/' + new_filekey )
        
        for row in csv_reader:
            if row:
                row = list(map(str.strip, row))
                if line_count == 0:
                    if 'ColToTruncate' in row:
                        colindex = row.index('ColToTruncate')
                        line_count += 1
                    else:
                        print('No ColToTruncate column found in '+ file_key)
                        return 'No ColToTruncate column found in '+ file_key
                else:
                    if len(row[colindex]) >= 10:
                        row[colindex] = row[colindex][0:2]
                    line_count += 1  
                content.append(row)
        
        # 用标准CSV工具生成内容，避免手动拼接格式错误
        output = io.StringIO()
        writer = csv.writer(output)
        writer.writerows(content)
        contentstring = output.getvalue()
        
        uploadByteStream = bytes(contentstring.encode('utf-8'))
        # 上传时添加已处理的元数据标记
        s3.put_object(
            Bucket=bucket, 
            Key=file_key, 
            Body=uploadByteStream,
            Metadata={'processed': 'true'}
        )
        return True
    except Exception as e:
        print(e)
        print('Error getting object {} from bucket {}. Make sure they exist and your bucket is in the same region as this function.'.format(key, bucket))
        raise e

额外优化说明 #

原代码中手动用, .join(row)拼接CSV内容会导致分隔符变成逗号加空格，不符合标准CSV格式，修改后用csv.writer生成内容，避免后续读取文件时出现格式错误。

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