从 Cloud Storage 进行事件驱动型传输

Storage Transfer Service 可以监听 Google Cloud 中的事件通知，以自动转移 Cloud Storage 存储桶中已添加或更新的数据。 详细了解事件驱动型转移的优势 。

来自 Cloud Storage 的事件驱动型转移使用 Pub/Sub 通知来了解何时修改或添加来源存储桶中的对象。系统不会检测到对象删除操作；在来源中删除对象不会删除目标存储桶中的关联对象。

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

配置权限

除了所有转移作业所需的 权限之外， 事件驱动型转移还需要 Pub/Sub Subscriber 角色。

  1. 找到项目的 Storage Transfer Service 服务代理的名称：

    1. 前往 googleServiceAccounts.get 参考页面。

      系统会打开标题为试用此方法的交互式面板。

    2. 在该面板的请求参数下，输入您的项目 ID。您在此处指定的项目必须是您用于管理 Storage Transfer Service 的项目，该项目可能与来源存储桶的项目不同。

    3. 点击执行。

    服务代理的电子邮件地址会作为 accountEmail 的值返回。复制此值。

    服务代理的电子邮件地址采用 project-PROJECT_NUMBER@storage-transfer-service.iam.gserviceaccount.com 格式。

  2. 将 Pub/Sub Subscriber 角色授予 Storage Transfer Service 服务代理。

    Cloud 控制台

    按照 通过控制台控制访问权限 Google Cloud 中的说明，向 Storage Transfer Service 服务授予 Pub/Sub Subscriber 角色。可以在主题、订阅或项目级授予该角色。

    gcloud CLI

    按照 设置政策中的说明 添加以下绑定：

    {
  "role": "roles/pubsub.subscriber",
  "members": [
    "serviceAccount:project-PROJECT_NUMBER@storage-transfer-service.iam.gserviceaccount.com"
}

配置 Pub/Sub

  1. 请确保您已满足将 Pub/Sub 与 Cloud Storage 搭配使用的 前提条件。

  2. 为来源 Cloud Storage 存储桶创建 Pub/Sub 通知。

    您无法使用 Google Cloud 控制台来管理 Pub/Sub 通知。 请改用 gcloud CLI 或 某一可用的客户端库 。

    gcloud storage buckets notifications create gs://SOURCE_BUCKET_NAME --topic=TOPIC_NAME

  3. 为该主题创建拉取订阅。您必须为每个转移作业创建单独的订阅。

    以下示例展示了用于创建拉取订阅的 Google Cloud CLI 命令。如需了解控制台说明和客户端库代码，请参阅 创建拉取订阅。

    gcloud pubsub subscriptions create SUBSCRIPTION_ID --topic=TOPIC_NAME --ack-deadline=300

创建转移作业

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

请勿在转移作业名称中包含敏感信息，例如个人身份信息 (PII) 或安全数据。资源名称可能会传播到其他 Google Cloud 资源的名称，并且可能会向您项目之外的 Google 内部系统公开。

Cloud 控制台

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

    前往创建转移作业

  2. 同时选择 Cloud Storage 作为来源和目标。

  3. 选择事件驱动型作为时间安排模式，然后点击下一步。

  4. 选择此转移的来源存储桶。

  5. 在事件流部分中，输入订阅名称：

    projects/PROJECT_NAME/subscriptions/SUBSCRIPTION_ID

  6. （可选）定义任何过滤条件，然后点击下一步 。

  7. 选择此转移的目标存储桶。

  8. （可选）输入转移的开始时间和结束时间。如果您未指定时间，转移作业将立即开始，并一直运行到 手动停止为止。

  9. 指定任何转移选项。如需了解详情，请参阅 创建转移作业页面。

  10. 点击创建 。

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

gcloud

如需使用 Google Cloud CLI 创建事件驱动型转移作业，请使用 gcloud transfer jobs create 命令和 --event-stream-name 标志：

gcloud transfer jobs create \
  gs://SOURCE_BUCKET \
  gs://DESTINATION_BUCKET \
  --event-stream-name=projects/PROJECT_NAME/subscriptions/SUBSCRIPTION_ID \
  --event-stream-starts=EVENT_STREAM_STARTS \
  --event-stream-expires=EVENT_STREAM_EXPIRES

将占位符替换为实际值：

  • SOURCE_BUCKET：来源 Cloud Storage 存储桶的名称。
  • DESTINATION_BUCKET：目标 Cloud Storage 存储桶。
  • PROJECT_NAME：包含 Pub/Sub 订阅的项目。
  • SUBSCRIPTION_ID：您创建的 Pub/Sub 订阅的 ID（例如 my-subscription）。
  • EVENT_STREAM_STARTS：使用 %Y-%m-%dT%H:%M:%S%z 日期时间格式（例如 2020-04-12T06:42:12+04:00）开始监听事件的时间。如果未设置，作业将在成功提交创建作业命令后开始运行并监听事件。
  • EVENT_STREAM_EXPIRES：停止监听事件的时间。如果未设置，作业将一直运行到手动停止为止。

如需查看受支持字段的完整列表，请参阅 gcloud transfer jobs create参考文档。

REST

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

transfer_job {
  "description": "YOUR DESCRIPTION",
  "status": "ENABLED",
  "projectId": "PROJECT_ID",
  "transferSpec" {
    "gcsDataSource" {
      "bucketName": "GCS_SOURCE_NAME"
    },
    "gcsDataSink": {
        "bucketName": "GCS_SINK_NAME"
    }
  }
  "eventStream" {
    "name": "projects/PROJECT_NAME/subscriptions/SUBSCRIPTION_ID",
    "eventStreamStartTime": "2022-12-02T01:00:00+00:00",
    "eventStreamExpirationTime": "2023-01-31T01:00:00+00:00"
  }
}

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

客户端库

Go

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

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


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

	// The name of the source GCS bucket.
	// gcsSourceBucket := "my-source-bucket"

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

	// The Pub/Sub topic to subscribe the event driven transfer to.
	// pubSubID := "projects/PROJECT_NAME/subscriptions/SUBSCRIPTION_ID"

	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_GcsDataSource{
					GcsDataSource: &storagetransferpb.GcsData{BucketName: gcsSourceBucket}},
				DataSink: &storagetransferpb.TransferSpec_GcsDataSink{
					GcsDataSink: &storagetransferpb.GcsData{BucketName: gcsSinkBucket}},
			},
			EventStream: &storagetransferpb.EventStream{Name: pubSubId},
			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", gcsSourceBucket, gcsSinkBucket, pubSubId, 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 CreateEventDrivenGcsTransfer {
  public static void main(String[] args) throws Exception {
    // Your Google Cloud Project ID