是否有一种方法可以通过最后一次使用气流来列出S3对象？

Question

到目前为止下面的代码

t1 = S3ListOperator(
    task_id='list_s3_files',
    bucket='mybucket',
    prefix='v01/{{ds}}/',
    delimiter='/'
)

然后使用S3CopyObjectOperator复制最新的文件。

Answer 1

不是一种特殊的“气流方式”，但您可以使用PythonOperator来完成这一任务。

all_objects = boto3.resource('s3').bucket(your_bucket_name).objects.iterator()
sorted_objs = sorted(all_objects, key=lambda o: o.last_modified)
latest_file = sorted_objs[-1]

虽然这不是一个“工业解决方案”，因为它需要提取所有的文件只是为了排序。S3不支持这样的“查询”。

如果您有一种可预测的方法来分割文件(例如，每天，每小时)，那么就不会那么糟糕了。