如何使用CDK将S3文件按文件名自动归类到对应子文件夹

我们通过CDK部署事件驱动的自动归类工作流，同时覆盖增量新上传文件和存量1TB CSV文件的批量处理，核心逻辑为解析文件名提取细粒度分类维度，自动生成对应层级目录完成文件迁移，全程保留原有日期分区规则。

• 环境准备
  首先初始化CDK项目，安装必要依赖，以Python CDK为例执行安装命令：
  pip install aws-cdk-lib aws-cdk.aws-lambda-python aws-cdk.aws-s3 aws-cdk.aws-s3-notifications

• 编写CDK资源栈代码

from aws_cdk import (
    Stack,
    aws_s3 as s3,
    aws_lambda as _lambda,
    aws_s3_notifications as s3_notif,
    Duration
)
from constructs import Construct

class FileAutoClassificationStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # 导入现有存储数据的S3桶，替换为实际桶名
        data_bucket = s3.Bucket.from_bucket_name(self, "ExistingDataBucket",
            bucket_name="your-existing-data-bucket"
        )

        # 定义文件归类处理Lambda
        classification_lambda = _lambda.Function(self, "FileClassificationLambda",
            runtime=_lambda.Runtime.PYTHON_3_10,
            handler="index.handler",
            code=_lambda.Code.from_asset("lambda"), # Lambda代码存放在项目根目录lambda文件夹下
            timeout=Duration.minutes(1),
            memory_size=256,
            environment={
                "TARGET_BUCKET": data_bucket.bucket_name,
                "FILENAME_DELIMITER": "_" # 按实际文件名分隔符调整
            }
        )

        # 授予Lambda桶读写权限
        data_bucket.grant_read_write(classification_lambda)

        # 配置S3事件通知，仅CSV文件上传时触发，处理增量数据
        data_bucket.add_event_notification(
            s3.EventType.OBJECT_CREATED,
            s3_notif.LambdaDestination(classification_lambda),
            s3.NotificationKeyFilter(suffix=".csv")
        )

• 编写Lambda处理逻辑
  在CDK项目根目录创建lambda/index.py文件，核心逻辑为解析文件名生成目标路径：

import boto3
import os

s3_client = boto3.client("s3")
TARGET_BUCKET = os.environ["TARGET_BUCKET"]
DELIMITER = os.environ["FILENAME_DELIMITER"]

def handler(event, context):
    for record in event["Records"]:
        source_bucket = record["s3"]["bucket"]["name"]
        source_key = record["s3"]["object"]["key"]
        
        # 跳过已完成归类的文件，避免循环触发
        if source_key.count("/") >= 4: # 按目标目录层级调整阈值
            return
        
        file_name = source_key.split("/")[-1]
        # 拆分文件名提取分类维度，示例规则：一级分类_二级分类_三级分类_日期.csv
        # 请按实际文件名规则修改拆分逻辑
        name_without_suffix = file_name.replace(".csv", "")
        name_parts = name_without_suffix.split(DELIMITER)
        
        if len(name_parts) < 4:
            # 不符合命名规则的文件归档到异常目录
            target_key = f"unclassified/{file_name}"
        else:
            level1, level2, level3, date_str = name_parts
            # 拼接细粒度目录，保留原有日期分区
            target_key = f"{level1}/{level2}/{level3}/{date_str}/{file_name}"
        
        # 迁移文件
        s3_client.copy_object(
            Bucket=TARGET_BUCKET,
            Key=target_key,
            CopySource={"Bucket": source_bucket, "Key": source_key}
        )
        # 如需保留原始文件可注释掉删除逻辑
        s3_client.delete_object(Bucket=source_bucket, Key=source_key)

• 存量数据批量处理
  处理现有1TB存量文件时，先注释掉CDK中S3事件通知的配置，部署完成后通过S3批处理操作批量调用Lambda处理所有存量CSV文件，处理完成后再开启事件通知接收增量数据。

• 文件名拆分逻辑需和现有所有文件的命名规则完全匹配，建议先拿10-20个不同命名的文件做测试，验证目录生成规则符合数据科学家的使用要求
• 若存在单个超过1GB的CSV文件，可将Lambda超时时间调整为5分钟、内存调整为512MB，避免迁移超时
• 建议开启S3版本控制，避免迁移过程中误删原始文件
• 异常目录的文件可定期巡检，补充适配特殊命名规则的解析逻辑

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