在 Amazon S3 中将数据重组为前缀

Question

我有一个用于大数据处理的 SQL 工具，可将数据转储到 .csv 文件中的 Amazon S3 存储桶中，前缀如下：

myS3Bucket
|
|__myTopPrefix
    |
    |_object1.csv
    |_object2.csv
    |_object3.csv
    |_object4.csv
    |_object5.csv
    |_object6.csv

我的要求是进一步将这些数据重组为具有以下结构的子前缀：

myS3Bucket
|
|__myTopPrefix
    |_subPrefix_1
    |   |_subSubPrefix_1
    |       |_object1.csv
    |       |_object2.csv
    |
    |_subPrefix_2
        |_subSubPrefix_2a
        |   |_object3.csv
        |   |_object4.csv
        |
        |_subSubPrefix_2b
            |_object5.csv
            |_object6.csv

通过查看该object.csv文件中的标题行来判断将哪个对象放在哪个前缀下。

我可以使用任何工具或（最好）AWS 服务来实现这一点，但我不知道该怎么做。我想过使用 aws lambda 函数来执行此操作并阅读 lambda 和其他资源的文档，但无法通过。我是 AWS 服务的新手，对它们还没有深入了解。

更新 1：object.csv 中的条目数可能以十亿为单位，例如 10-120 亿。

更新 2：我改变了服务方式，现在“更新 1”不再有效。任何 object.csv 中的条目现在大约为 10,000，但在一次 SQL 脚本执行中将转储到 S3 中的目标文件现在将增加到 12。在最坏的情况下，典型文件大小将是 150-200 MB（这永远不会会发生的XD）

此外，我需要在将每个新对象推送到 S3 之后进行对象重组。还有我已经创建的前缀名称（或分层路径），但它们最初不包含任何数据/对象。

作为一个工作流程，我在我的工具中处理数据，然后将其转储到S3://myS3Bucket/myTopPrefix/ 现在，该数据应该根据文件头中的信息移动到适当的前缀。任务完成！

更新 3： 对于那些正在考虑如何从单个 SQL 脚本生成 12 个对象的人：基本上在这个工具中，我可以编写多个 select ... from ... where ...，每个查询将输出数据作为一个硬编码的对象（工具的限制）在S3://myS3Bucket/myTopPrefix/ 下。在当前的实现中，单次执行 SQL 脚本将生成 12 个对象。

非常感谢任何帮助。 谢谢！

Answer 1

假设：

• 您希望在创建对象时移动对象（无需移动历史对象）
• 目标路径的名称可以从每个文件的第一行推导出来
• 输入路径与输出路径完全分开（以避免导致无限循环）。例如，传入的文件来自input/，然后被移动到不以input/开头的路径。
• Amazon S3 已配置为在输入路径中创建新对象时触发 AWS Lambda 函数

另外，请注意 Amazon S3 没有“移动”命令。相反，需要复制然后删除对象。

import boto3
import urllib

def lambda_handler(event, context):
   
    s3_client = boto3.client('s3')
    
    bucket = event['Records'][0]['s3']['bucket']['name']
    key = urllib.parse.unquote_plus(event['Records'][0]['s3']['object']['key'])
    
    # Read the first line of the object
    # Uses iter_lines() to give us the first line from the StreamingBody without downloading the whole object
    object = s3_client.get_object(Bucket=bucket, Key=key)
    for line in object['Body'].iter_lines():
        line1 = line.decode('utf-8')
        
        # Extract information from first line here (this is just dummy code, put your code here)
        # Note that the directory does not need to exist already -- S3 does not need folders created
        destination_path = 'foo/bar/'
        
        # Break to exit loop, since we only want first line
        break
    
    # Copy object to new path
    filename = key[key.rfind('/')+1 : ] # Extract name after last slash
    destination_key = destination_path + filename
    
    s3_client.copy_object(
        Bucket=bucket,
        Key=destination_key,
        CopySource=f'{bucket}/{key}'
        )
    
    # Delete incoming object
    s3_client.delete_object(Bucket=bucket, Key=key)