从 Google Cloud Storage 存储桶复制到 S3 存储桶

Question

我设置了一个气流工作流程，将一些文件从 s3 摄取到 Google Cloud 存储，然后运行 ​​sql 查询工作流程以在 Big Query 上创建新表。在工作流结束时，我需要将最后一个 Big Query 表的输出推送到 Google Cloud Storage，然后从那里推送到 S3。

我已经使用BigQueryToCloudStorageOperator python 运算符破解了大查询表到谷歌云存储的传输，没有任何问题。但是，从 Google Cloud Storage 到 S3 的转移似乎是一条不太受欢迎的路线，我一直无法找到可以在我的 Airflow 工作流程中自动化的解决方案。

我知道rsync 是gsutil 的一部分，并且已经开始工作（参见帖子Exporting data from Google Cloud Storage to Amazon S3），但我无法将其添加到我的工作流程中。

我有一个在计算引擎实例上运行的 dockerised 气流容器。

非常感谢帮助解决这个问题。

非常感谢！

Answer 1

所以我们也使用rsync 在 S3 和 GCS 之间移动数据，

您首先需要让 bash 脚本正常工作，例如 gsutil -m rsync -d -r gs://bucket/key s3://bucket/key

对于 s3，您还需要提供 AWS_ACCESS_KEY_ID 和 AWS_SECRET_ACCESS_KEY 作为环境变量。

然后定义您的 BashOperator 并将其放入您的 DAG 文件中

rsync_yesterday = BashOperator(task_id='rsync_task_' + table,
                                bash_command='Your rsync script',
                                dag=dag)

Answer 2

Google 建议使用 transfer service 在云平台之间进行传输。您可以使用他们的 python API 以编程方式设置传输。通过这种方式，数据直接在 S3 和谷歌云存储之间传输。使用gsutil 和rsync 的缺点是数据必须通过执行rsync 命令的机器/实例。这可能是一个瓶颈。

Google Cloud Storage Transfer Service Doc

Answer 3

我需要使用 AWS Lambda 将对象从 GC 存储桶复制到 S3。

Python boto3 库允许从 GC 存储桶中列出和下载对象。

以下是将“sample-data-s3.csv”对象从 GC 存储桶复制到 s3 存储桶的示例 lambda 代码。

import boto3
import io

s3 = boto3.resource('s3')

google_access_key_id="GOOG1EIxxMYKEYxxMQ"
google_access_key_secret="QifDxxMYSECRETKEYxxVU1oad1b"

gc_bucket_name="my_gc_bucket"


def get_gcs_objects(google_access_key_id, google_access_key_secret,
                     gc_bucket_name):
    """Gets GCS objects using boto3 SDK"""
    client = boto3.client("s3", region_name="auto",
                          endpoint_url="https://storage.googleapis.com",
                          aws_access_key_id=google_access_key_id,
                          aws_secret_access_key=google_access_key_secret)

    # Call GCS to list objects in gc_bucket_name
    response = client.list_objects(Bucket=gc_bucket_name)

    # Print object names
    print("Objects:")
    for blob in response["Contents"]:
        print(blob)    

    object = s3.Object('my_aws_s3_bucket', 'sample-data-s3.csv')
    f = io.BytesIO()
    client.download_fileobj(gc_bucket_name,"sample-data.csv",f)
    object.put(Body=f.getvalue())

def lambda_handler(event, context):
    get_gcs_objects(google_access_key_id,google_access_key_secret,gc_bucket_name) 

您可以循环通过blob从GC桶中下载所有对象。

希望这对想要使用 AWS lambda 将对象从 GC 存储桶传输到 s3 存储桶的人有所帮助。

Answer 4

最简单的整体选项是gsutil rsync，但是在某些情况下 rsync 可能会占用太多资源或不够快。

结合其他选择：

• 查看亚马逊的数据传输服务，称为Amazon S3 Transfer Acceleration，它允许您将数据从 GCS 导入 S3
• 使用 Hadoop DistCp 和 DataProc 来并行化复制过程 (see example)