Google uses AI technology to translate content into your preferred language. AI translations can contain errors.

从 AWS S3 进行事件驱动型转移

Storage Transfer Service 可以侦听 AWS 中的事件通知，以自动将来源位置中添加或更新的数据转移到 Cloud Storage 存储桶中。详细了解事件驱动型转移的优势。

事件驱动型转移会侦听 Amazon S3 事件通知发送到 Amazon SQS，以了解源存储桶中的对象何时已修改或添加。系统不会检测到对象删除操作；在来源中删除对象不会删除目标存储桶中的关联对象。

事件驱动型转移始终使用 Cloud Storage 存储桶作为目标。

准备工作

按照说明授予对目标 Cloud Storage 存储桶的必要权限：

配置对目标的访问权限：Cloud Storage

创建 SQS 队列

在 AWS 控制台中，转到 Simple Queue Service 页面。
点击创建队列 。
为此队列输入名称。

在访问权限政策 部分中，选择高级。此时会显示一个 JSON 对象。

标准 AWS 区域

{
  "Version": "2008-10-17",
  "Id": "\_\_default\_policy\_ID",
  "Statement": [
    {
      "Sid": "\_\_owner\_statement",
      "Effect": "Allow",
      "Principal": {
        "AWS": "01234567890"
      },
      "Action": [
        "SQS:*"
      ],
      "Resource": "arn:aws:sqs:us-west-2:01234567890:test"
    }
  ]
}

AWS GovCloud 区域

{
  "Version": "2008-10-17",
  "Id": "\_\_default\_policy\_ID",
  "Statement": [
    {
      "Sid": "\_\_owner\_statement",
      "Effect": "Allow",
      "Principal": {
        "AWS": "01234567890"
      },
      "Action": [
        "SQS:*"
      ],
      "Resource": "arn:aws-us-gov:sqs:us-gov-west-1:01234567890:test"
    }
  ]
}

复制 AWS 和 Resource 的值。这些值对于每个项目都是唯一的。

将上一步中的特定 AWS 和 Resource 值粘贴到以下 JSON 代码段中：

标准 AWS 区域

{
"Version": "2012-10-17",
"Id": "example-ID",
"Statement": [
  {
    "Sid": "example-statement-ID",
    "Effect": "Allow",
    "Principal": {
      "Service": "s3.amazonaws.com"
    },
    "Action": "SQS:SendMessage",
    "Resource": "RESOURCE",
    "Condition": {
      "StringEquals": {
        "aws:SourceAccount": "AWS"
      },
      "ArnLike": {
        "aws:SourceArn": "arn:aws:s3:::S3_BUCKET_NAME"
      }
    }
  }
]
}

AWS GovCloud 区域

{
"Version": "2012-10-17",
"Id": "example-ID",
"Statement": [
  {
    "Sid": "example-statement-ID",
    "Effect": "Allow",
    "Principal": {
      "Service": "s3.amazonaws.com"
    },
    "Action": "SQS:SendMessage",
    "Resource": "RESOURCE",
    "Condition": {
      "StringEquals": {
        "aws:SourceAccount": "AWS"
      },
      "ArnLike": {
        "aws:SourceArn": "arn:aws-us-gov:s3:::S3_BUCKET_NAME"
      }
    }
  }
]
}

将 S3_BUCKET_NAME 替换为 S3 来源存储桶的名称。

复制此已完成的 JSON 代码段，并使用它替换访问权限政策 部分中显示的 JSON。
点击创建队列 。

完成后，记下队列的 Amazon 资源名称 (ARN)。

在 S3 存储桶上启用通知

在 AWS 控制台中，转到 S3 页面。
在存储桶列表中，选择您的源存储桶。
选择属性标签页。
在事件通知 部分，点击创建事件通知 。
为此事件指定一个名称。
在事件类型 部分，选择所有对象创建事件 。
选择 SQS 队列 作为目标，然后选择为此转移作业创建的队列。
点击保存更改。

配置权限

按照 “配置对来源的访问权限：Amazon S3” 中的说明创建访问密钥 ID 和密钥或 Federated Identity 角色。

按照说明操作时，请在系统要求您指定自定义角色或自定义信任政策时使用以下 JSON：

标准 AWS 区域

{
  "Version": "2012-10-17",
  "Statement": [
      {
          "Effect": "Allow",
          "Action": [
              "sqs:DeleteMessage",
              "sqs:ChangeMessageVisibility",
              "sqs:ReceiveMessage",
              "s3:GetObject",
              "s3:ListBucket"
          ],
          "Resource": [
              "arn:aws:s3:::S3_BUCKET_NAME",
              "arn:aws:s3:::S3_BUCKET_NAME/*",
              "AWS_QUEUE_ARN"
          ]
      }
  ]
}

创建后，请记下以下信息：

对于用户，请记下访问密钥 ID 和密钥。
对于 Federated Identity 角色，请记下 Amazon 资源名称 (ARN)，其格式如下： arn:aws:iam::AWS_ACCOUNT:role/AWS_ROLE_NAME

AWS GovCloud 区域

{
  "Version": "2012-10-17",
  "Statement": [
      {
          "Effect": "Allow",
          "Action": [
              "sqs:DeleteMessage",
              "sqs:ChangeMessageVisibility",
              "sqs:ReceiveMessage",
              "s3:GetObject",
              "s3:ListBucket"
          ],
          "Resource": [
              "arn:aws-us-gov:s3:::S3_BUCKET_NAME",
              "arn:aws-us-gov:s3:::S3_BUCKET_NAME/*",
              "AWS_QUEUE_ARN"
          ]
      }
  ]
}

创建后，请记下以下信息：

对于用户，请记下访问密钥 ID 和密钥。
对于 Federated Identity 角色，请记下 Amazon 资源名称 (ARN)，其格式如下：arn:aws-us-gov:iam::AWS_ACCOUNT:role/AWS_ROLE_NAME

创建转移作业

您可以使用 REST API 或 Google Cloud 控制台创建基于事件的转移作业。

Cloud 控制台

进入 Google Cloud 控制台中的创建转移作业 页面。

进入创建转移作业
选择 Amazon S3 作为来源类型，并选择 Cloud Storage 作为目标。
选择事件驱动型作为时间安排模式，然后点击下一步。
输入您的 S3 存储桶名称。存储分区名称是其在 AWS 管理控制台中显示的名称。例如 my-aws-bucket。
选择身份验证方法，然后输入您在上一部分中创建并记下的所需信息。
输入您之前创建的 Amazon SQS 队列 ARN。它采用以下格式之一：
- 对于标准 AWS 区域：arn:aws:sqs:AWS_REGION:AWS_ACCOUNT:AWS_QUEUE_NAME
- 对于 AWS GovCloud 区域：arn:aws-us-gov:sqs:AWS_REGION:AWS_ACCOUNT:AWS_QUEUE_NAME
（可选）定义任何过滤条件，然后点击下一步 。
选择目标 Cloud Storage 存储桶以及可选的路径。
（可选）输入转移的开始时间和结束时间。如果您未指定时间，转移作业将立即开始，并一直运行到手动停止为止。
指定任何转移选项。如需了解详情，请参阅创建转移作业页面。
点击创建。

创建后，转移作业将开始运行，且事件监听器会等待 SQS 队列上的通知。作业详情页面每小时显示一次操作，并包含每项作业转移的数据的详细信息。

REST

如需使用 REST API 创建事件驱动型转移作业，请将以下 JSON 对象发送到 transferJobs.create 端点：

{
"description": "DESCRIPTION",
"status": "ENABLED",
"projectId": "PROJECT_ID",
"transferSpec": {
  "awsS3DataSource": {
    "bucketName": "S3_BUCKET_NAME",
    "roleArn": "AWS_ROLE_ARN"
  },
  "gcsDataSink": {
    "bucketName": "GCS_BUCKET_NAME"
  }
},
"eventStream": {
  "name": "AWS_QUEUE_ARN",
  "eventStreamStartTime": "2022-12-02T01:00:00+00:00",
  "eventStreamExpirationTime": "2023-01-31T01:00:00+00:00"
}
}

上述 JSON 中的占位符使用以下值：

DESCRIPTION 是转移作业的说明。
PROJECT_ID 是创建转移作业的 Google Cloud 云项目的 ID。
S3_BUCKET_NAME 是 Amazon S3 来源存储桶的名称。
AWS_ROLE_ARN 是您创建的 Federated Identity 角色的 ARN。例如，对于标准 AWS 区域，为 arn:aws:iam::1234567891011:role/aws-role-name；对于 AWS GovCloud 区域，为 arn:aws-us-gov:iam::1234567891011:role/aws-role-name。
GCS_BUCKET_NAME 是 Cloud Storage 目标存储桶的名称。
AWS_QUEUE_ARN 是 SQS 队列的 ARN。例如，对于标准 AWS 区域，为 arn:aws:sqs:us-east-1:1234567891011:s3-notification-queue；对于 AWS GovCloud 区域，为 arn:aws-us-gov:sqs:us-gov-east-1:1234567890:event-queue。

eventStreamStartTime 和 eventStreamExpirationTime 是可选的。如果省略了开始时间，则转移作业会立即开始；如果省略了结束时间，则转移作业会一直持续，直到手动停止为止。

客户端库

Go

如需了解如何安装和使用 Storage Transfer Service 客户端库，请参阅 Storage Transfer Service 客户端库。如需了解详情，请参阅 Storage Transfer Service Go API 参考文档。

如需向 Storage Transfer Service 进行身份验证，请设置应用默认凭据。如需了解详情，请参阅为本地开发环境设置身份验证。


func createEventDrivenAWSTransfer(w io.Writer, projectID string, s3SourceBucket string, gcsSinkBucket string, sqsQueueARN string) (*storagetransferpb.TransferJob, error) {
	// Your Google Cloud Project ID.
	// projectID := "my-project-id"

	// The name of the source AWS S3 bucket.
	// s3SourceBucket := "my-source-bucket"

	// The name of the GCS bucket to transfer objects to.
	// gcsSinkBucket := "my-sink-bucket"

	// The Amazon Resource Name (ARN) of the AWS SNS queue to subscribe the event driven transfer to.
	// sqsQueueARN := "arn:aws:sqs:us-east-1:1234567891011:s3-notification-queue"

	// The AWS access key credential, should be accessed via environment variable for security
	awsAccessKeyID := os.Getenv("AWS_ACCESS_KEY_ID")

	// The AWS secret key credential, should be accessed via environment variable for security
	awsSecretKey := os.Getenv("AWS_SECRET_ACCESS_KEY")

	ctx := context.Background()
	client, err := storagetransfer.NewClient(ctx)
	if err != nil {
		return nil, fmt.Errorf("storagetransfer.NewClient: %w", err)
	}
	defer client.Close()

	req := &storagetransferpb.CreateTransferJobRequest{
		TransferJob: &storagetransferpb.TransferJob{
			ProjectId: projectID,
			TransferSpec: &storagetransferpb.TransferSpec{
				DataSource: &storagetransferpb.TransferSpec_AwsS3DataSource{
					AwsS3DataSource: &storagetransferpb.AwsS3Data{
						BucketName: s3SourceBucket,
						AwsAccessKey: &storagetransferpb.AwsAccessKey{
							AccessKeyId:     awsAccessKeyID,
							SecretAccessKey: awsSecretKey,
						}},
				},
				DataSink: &storagetransferpb.TransferSpec_GcsDataSink{
					GcsDataSink: &storagetransferpb.GcsData{BucketName: gcsSinkBucket}},
			},
			EventStream: &storagetransferpb.EventStream{Name: sqsQueueARN},
			Status:      storagetransferpb.TransferJob_ENABLED,
		},
	}
	resp, err := client.CreateTransferJob(ctx, req)
	if err != nil {
		return nil, fmt.Errorf("failed to create transfer job: %w", err)
	}

	fmt.Fprintf(w, "Created an event driven transfer job from %v to %v subscribed to %v with name %v", s3SourceBucket, gcsSinkBucket, sqsQueueARN, resp.Name)
	return resp, nil
}

Java

如需了解如何安装和使用 Storage Transfer Service 客户端库，请参阅 Storage Transfer Service 客户端库。如需了解详情，请参阅 Storage Transfer Service Java API 参考文档。

如需向 Storage Transfer Service 进行身份验证，请设置应用默认凭据。如需了解详情，请参阅为本地开发环境设置身份验证。


import com.google.storagetransfer.v1.proto.StorageTransferServiceClient;
import com.google.storagetransfer.v1.proto.TransferProto;
import com.google.storagetransfer.v1.proto.TransferTypes;

public class CreateEventDrivenAwsTransfer {
  public static void main(String[] args) throws Exception {
    // Your Google Cloud Project ID
    String projectId = "your-project-id";

    // The name of the source AWS bucket to transfer data from
    String s3SourceBucket = "yourS3SourceBucket";

    // The name of the GCS bucket to transfer data to
    String gcsSinkBucket = "your-gcs-bucket";

    // The ARN of the SQS queue to subscribe to
    String sqsQueueArn = "arn:aws:sqs:us-east-1:1234567891011:s3-notification-queue";

    createEventDrivenAwsTransfer(projectId, s3SourceBucket, gcsSinkBucket, sqsQueueArn);
  }

  public static void createEventDrivenAwsTransfer(
      String projectId, String s3SourceBucket, String gcsSinkBucket, String sqsQueueArn)
      throws Exception {
    // Initialize client that will be used to send requests. This client only needs to be created
    // once, and can be reused for multiple requests. After completing all of your requests, call
    // the "close" method on the client to safely clean up any remaining background resources,
    // or use "try-with-close" statement to do this automatically.
    try (StorageTransferServiceClient storageTransfer = StorageTransferServiceClient.create()) {

      // The ID used to access your AWS account. Should be accessed via environment variable.
      String awsAccessKeyId = System.getenv("AWS_ACCESS_KEY_ID");

      // The Secret Key used to access your AWS account. Should be accessed via environment
      // variable.
      String awsSecretAccessKey = System.getenv("AWS_SECRET_ACCESS_KEY");

      TransferTypes.TransferJob transferJob =
          TransferTypes.TransferJob.newBuilder()
              .setProjectId(projectId)
              .setTransferSpec(
                  TransferTypes.TransferSpec.newBuilder()
                      .setAwsS3DataSource(
                          TransferTypes.AwsS3Data.newBuilder()
                              .setBucketName(s3SourceBucket)
                              .setAwsAccessKey(
                                  TransferTypes.AwsAccessKey.newBuilder()
                                      .setAccessKeyId(awsAccessKeyId)
                                      .setSecretAccessKey(awsSecretAccessKey))
                              .build())
                      .setGcsDataSink(
                          TransferTypes.GcsData.newBuilder().setBucketName(gcsSinkBucket)))
              .setStatus(TransferTypes.TransferJob.Status.ENABLED)
              .setEventStream(TransferTypes.EventStream.newBuilder().setName(sqsQueueArn).build())
              .build();

      TransferTypes.TransferJob response =
          storageTransfer.createTransferJob(
              TransferProto.CreateTransferJobRequest.newBuilder()
                  .setTransferJob(transferJob)
                  .build());

      System.out.println(
          "Created a transfer job from "
              + s3SourceBucket
              + " to "
              + gcsSinkBucket
              + " subscribed to "
              + sqsQueueArn
              + " with name "
              + response.getName());
    }
  }
}

Node.js

如需了解如何安装和使用 Storage Transfer Service 客户端库，请参阅 Storage Transfer Service 客户端库。如需了解详情，请参阅 Storage Transfer Service Node.js API 参考文档。

如需向 Storage Transfer Service 进行身份验证，请设置应用默认凭据。如需了解详情，请参阅为本地开发环境设置身份验证。


// Imports the Google Cloud client library
const {
  StorageTransferServiceClient,
} = require('@google-cloud/storage-transfer');

/**
 * TODO(developer): Uncomment the following lines before running the sample.
 */
// The ID of the Google Cloud Platform Project that owns the job
// projectId = 'my-project-id'

// AWS S3 source bucket name
// s3SourceBucket = 'my-s3-source-bucket'

// Google Cloud Storage destination bucket name
// gcsSinkBucket = 'my-gcs-destination-bucket'

// The ARN of the SQS queue to subscribe to
// sqsQueueArn = 'arn:aws:sqs:us-east-1:1234567891011:s3-notification-queue'

// AWS Access Key ID. Should be accessed via environment variable for security.
// awsAccessKeyId = 'AKIA...'

// AWS Secret Access Key. Should be accessed via environment variable for security.
// awsSecretAccessKey = 'HEAoMK2.../...ku8'

// Creates a client
const client = new StorageTransferServiceClient();

/**
 * Creates an event driven transfer that tracks an SQS queue.
 */
async function createEventDrivenAwsTransfer() {
  const [transferJob] = await client.createTransferJob({
    transferJob: {
      projectId,
      status: 'ENABLED',
      transferSpec: {
        awsS3DataSource: {
          bucketName: s3SourceBucket,
          awsAccessKey: {
            accessKeyId: awsAccessKeyId,
            secretAccessKey: awsSecretAccessKey,
          },
        },
        gcsDataSink: {
          bucketName: gcsSinkBucket,
        },
      },
      eventStream: {
        name: sqsQueueArn,
      },
    },
  });

  console.log(
    `Created an event driven transfer from '${s3SourceBucket}' to '${gcsSinkBucket}' with name ${transferJob.name}`
  );
}

createEventDrivenAwsTransfer();

Python

如需了解如何安装和使用 Storage Transfer Service 客户端库，请参阅 Storage Transfer Service 客户端库。如需了解详情，请参阅 Storage Transfer Service Python API 参考文档。

如需向 Storage Transfer Service 进行身份验证，请设置应用默认凭据。如需了解详情，请参阅为本地开发环境设置身份验证。


from google.cloud import storage_transfer


def create_event_driven_aws_transfer(
    project_id: str,
    description: str,
    source_s3_bucket: str,
    sink_gcs_bucket: str,
    sqs_queue_arn: str,
    aws_access_key_id: str,
    aws_secret_access_key: str,
):
    """Create an event driven transfer between two GCS buckets that tracks an AWS SQS queue"""

    client = storage_transfer.StorageTransferServiceClient()

    # The ID of the Google Cloud Platform Project that owns the job
    # project_id = 'my-project-id'

    # A description of this job
    # description = 'Creates an event-driven transfer that tracks an SQS queue'

    # AWS S3 source bucket name
    # source_s3_bucket = 'my-s3-source-bucket'

    # Google Cloud Storage destination bucket name
    # sink_gcs_bucket = 'my-gcs-destination-bucket'

    # The ARN of the SQS queue to subscribe to
    # pubsub_id = 'arn:aws:sqs:us-east-1:1234567891011:s3-notification-queue'

    # AWS Access Key ID. Should be accessed via environment variable for security purposes.
    # aws_access_key_id = 'AKIA...'

    # AWS Secret Access Key. Should be accessed via environment variable for security purposes.
    # aws_secret_access_key = 'HEAoMK2.../...ku8'

    transfer_job_request = storage_transfer.CreateTransferJobRequest(
        {
            "transfer_job": {
                "project_id": project_id,
                "description": description,
                "status": storage_transfer.TransferJob.Status.ENABLED,
                "transfer_spec": {
                    "aws_s3_data_source": {
                        "bucket_name": source_s3_bucket,
                        "aws_access_key": {
                            "access_key_id": aws_access_key_id,
                            "secret_access_key": aws_secret_access_key,
                        },
                    },
                    "gcs_data_sink": {
                        "bucket_name": sink_gcs_bucket,
                    },
                },
                "event_stream": {
                    "name": sqs_queue_arn,
                },
            },
        }
    )

    result = client.create_transfer_job(transfer_job_request)
    print(f"Created transferJob: {result.name}")

从 AWS S3 进行事件驱动型转移 使用集合让一切井井有条 根据您的偏好保存内容并对其进行分类。

准备工作

创建 SQS 队列

标准 AWS 区域

AWS GovCloud 区域

标准 AWS 区域

AWS GovCloud 区域

在 S3 存储桶上启用通知

配置权限

标准 AWS 区域

AWS GovCloud 区域

创建转移作业

Cloud 控制台

REST

客户端库

Go

Java

Node.js

Python

从 AWS S3 进行事件驱动型转移