python - 如何在 Airflow 中使用 s3 hook

Question

我有一个 s3 文件夹位置，我要移动到 GCS。
我正在使用 Airflow 使运动发生。

在这种环境下，我的 s3 是一个“不断增长”的文件夹，这意味着我们在获取文件后不会删除它们。

def GetFiles(**kwargs):
    foundfiles = False

    s3 = S3Hook(aws_conn_id='S3_BDEX')
    s3.get_conn()
    bucket = s3.get_bucket(
        bucket_name='/file.share.external.bdex.com/Offrs'
    )
    files = s3.list_prefixes(bucket_name='/file.share.external.bdex.com/Offrs')
    print("BUCKET:  {}".format(files))


check_for_file = BranchPythonOperator(
    task_id='Check_FTP_and_Download',
    provide_context=True,
    python_callable=GetFiles,
    dag=dag
)

我在这里需要的是文件列表及其创建日期/时间。通过这种方式，我可以比较现有文件以确定它们是否是新文件。

我知道我可以连接，因为函数 get_bucket功能起作用了。
但是，在这种情况下，我收到以下错误:

Invalid bucket name "/file.share.external.bdex.com/Offrs": Bucket name must match the regex "^[a-zA-Z0-9.\-_]{1,255}$"

谢谢

Answer 1

存储桶名称错误。如果 url 是 s3://something/path/to/file，则存储桶名称是“something”。