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

Cloud Storage からのイベントドリブン転送

Storage Transfer Service は、Google Cloud のイベント通知をリッスンし、Cloud Storage バケットで追加または更新されたデータを自動的に転送できます。イベントドリブン転送のメリットの詳細をご覧ください。

Cloud Storage からのイベントドリブン転送では、Pub/Sub 通知を使用して、転送元バケット内のオブジェクトが変更または追加されたタイミングを確認します。オブジェクトの削除は検出されません。転送元でオブジェクトを削除しても、転送先バケット内の関連オブジェクトは削除されません。

イベントドリブン転送では、常に Cloud Storage バケットが転送先として使用されます。

権限を構成する

すべての転送ジョブに必要な権限に加えて、イベントドリブン転送には Pub/Sub Subscriber ロールが必要です。

プロジェクトの Storage Transfer Service サービスエージェントの名前を確認します。
1. googleServiceAccounts.get リファレンスページに移動します。
  
  [Try this method] というインタラクティブパネルが開きます。
2. パネルの [Request parameters] にプロジェクト ID を入力します。ここで指定するプロジェクトは Storage Transfer Service の管理に使用しているプロジェクトで、これは、ソースバケットのプロジェクトとは異なる場合があります。
3. [実行] をクリックします。
サービスエージェントのメールアドレスが accountEmail の値として返されます。この値をコピーします。

サービスエージェントのメールアドレスは project-PROJECT_NUMBER@storage-transfer-service.iam.gserviceaccount.com の形式です。
Pub/Sub Subscriber ロールを Storage Transfer Service サービスエージェントに付与します。
Cloud コンソール
コンソールによるアクセス制御の手順に沿って、Storage Transfer Service サービスに Pub/Sub Subscriber ロールを付与します。 Google Cloud このロールは、トピック、サブスクリプション、プロジェクトレベルで付与できます。
gcloud CLI
ポリシーの設定の手順で、次のバインディングを追加します。
{ "role": "roles/pubsub.subscriber", "members": [ "serviceAccount:project-PROJECT_NUMBER@storage-transfer-service.iam.gserviceaccount.com" }

Pub/Sub を構成する

Cloud Storage で Pub/Sub の前提条件を満たしていることを確認します。
転送元の Cloud Storage バケットの Pub/Sub 通知を作成します。

コンソールでは Pub/Sub 通知を管理できません。 Google Cloud 代わりに gcloud CLI または使用可能なクライアントライブラリのいずれかを使用してください。
```
gcloud storage buckets notifications create gs://SOURCE_BUCKET_NAME --topic=TOPIC_NAME
```
トピックに pull サブスクリプションを作成します。転送ジョブごとに個別のサブスクリプションを作成する必要があります。

次の例は、pull サブスクリプションを作成するための Google Cloud CLI コマンドを示しています。コンソールの手順とクライアントライブラリのコードについては、 pull サブスクリプションを作成するをご覧ください。
```
gcloud pubsub subscriptions create SUBSCRIPTION_ID --topic=TOPIC_NAME --ack-deadline=300
```

転送ジョブを作成する

コンソール、Google Cloud CLI、または REST API を使用して、イベントベースの転送ジョブを作成できます。 Google Cloud

転送ジョブ名に、個人を特定できる情報（PII）やセキュリティデータなどの機密情報を含めないでください。リソース名は、他の Google Cloud リソースの名前に反映され、プロジェクト外部の Google 内部システムに公開される場合があります。

Cloud コンソール

[転送ジョブを作成] ページに移動します。 Google Cloud コンソール。

[転送ジョブを作成] に移動
ソースと宛先の両方で「Cloud Storage」を選択します。
[スケジュールモード] として [イベントドリブン] を選択し、[次のステップ] をクリックします。
この転送の転送元バケットを選択します。
[イベントストリーム] セクションで、サブスクリプション名を入力します。

projects/PROJECT_NAME/subscriptions/SUBSCRIPTION_ID
必要に応じてフィルタを定義し、[次のステップ] をクリックします。
この転送の転送先バケットを選択します。
必要に応じて、転送の開始時刻と終了時刻を入力します。時間を指定しない場合、転送はすぐに開始し、手動で停止するまで実行されます。
転送オプションを指定します。詳細については、転送の作成ページをご覧ください。
[作成] をクリックします。

作成されると、転送ジョブの実行が開始し、イベントリスナーが Pub/Sub サブスクリプションの通知を待機します。ジョブの詳細ページには、1 時間ごとにオペレーションが 1 つ表示されます。このページには、各ジョブで転送されたデータの詳細も表示されます。

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 のリファレンスドキュメントをご覧ください。


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
    String projectId = "your-project-id";

    // The name of the GCS AWS bucket to transfer data from
    String gcsSourceBucket = "your-gcs-source-bucket";

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

    // The ARN of the PubSub queue to subscribe to
    String sqsQueueArn = "projects/PROJECT_NAME/subscriptions/SUBSCRIPTION_ID";

    createEventDrivenGcsTransfer(projectId, gcsSourceBucket, gcsSinkBucket, sqsQueueArn);
  }

  public static void createEventDrivenGcsTransfer(
      String projectId, String gcsSourceBucket, String gcsSinkBucket, String pubSubId)
      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()) {

      TransferTypes.TransferJob transferJob =
          TransferTypes.TransferJob.newBuilder()
              .setProjectId(projectId)
              .setTransferSpec(
                  TransferTypes.TransferSpec.newBuilder()
                      .setGcsDataSource(
                          TransferTypes.GcsData.newBuilder().setBucketName(gcsSourceBucket))
                      .setGcsDataSink(
                          TransferTypes.GcsData.newBuilder().setBucketName(gcsSinkBucket)))
              .setStatus(TransferTypes.TransferJob.Status.ENABLED)
              .setEventStream(TransferTypes.EventStream.newBuilder().setName(pubSubId).build())
              .build();

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

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

Node.js

Storage Transfer Service 用のクライアントライブラリをインストールして使用する方法については、 Storage Transfer Service のクライアントライブラリをご覧ください。詳細については、Storage Transfer Service Node.js API のリファレンスドキュメントをご覧ください。


// 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'

// Google Cloud Storage source bucket name
// gcsSourceBucket = 'my-gcs-source-bucket'

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

// The subscription ID to a Pubsub queue to track
// pubsubId = 'projects/PROJECT_NAME/subscriptions/SUBSCRIPTION_ID'

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

/**
 * Creates an event driven transfer that tracks a Pubsub subscription.
 */
async function createEventDrivenGcsTransfer() {
  const [transferJob] = await client.createTransferJob({
    transferJob: {
      projectId,
      status: 'ENABLED',
      transferSpec: {
        gcsDataSource: {
          bucketName: gcsSourceBucket,
        },
        gcsDataSink: {
          bucketName: gcsSinkBucket,
        },
      },
      eventStream: {
        name: pubsubId,
      },
    },
  });

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

createEventDrivenGcsTransfer();

Python

Storage Transfer Service 用のクライアントライブラリをインストールして使用する方法については、 Storage Transfer Service のクライアントライブラリをご覧ください。詳細については、Storage Transfer Service Python API のリファレンスドキュメントをご覧ください。


from google.cloud import storage_transfer


def create_event_driven_gcs_transfer(
    project_id: str,
    description: str,
    source_bucket: str,
    sink_bucket: str,
    pubsub_id: str,
):
    """Create an event driven transfer between two GCS buckets that tracks a PubSub subscription"""

    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 a pubsub subscription'

    # Google Cloud Storage source bucket name
    # source_bucket = 'my-gcs-source-bucket'

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

    # The Pubsub Subscription ID to track
    # pubsub_id = 'projects/PROJECT_NAME/subscriptions/SUBSCRIPTION_ID'

    transfer_job_request = storage_transfer.CreateTransferJobRequest(
        {
            "transfer_job": {
                "project_id": project_id,
                "description": description,
                "status": storage_transfer.TransferJob.Status.ENABLED,
                "transfer_spec": {
                    "gcs_data_source": {
                        "bucket_name": source_bucket,
                    },
                    "gcs_data_sink": {
                        "bucket_name": sink_bucket,
                    },
                },
                "event_stream": {
                    "name": pubsub_id,
                },
            },
        }
    )

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

イベントドリブン転送をモニタリングする

イベントドリブン転送を作成すると、Storage Transfer Service によって転送ジョブが作成されます。開始時刻に達すると、転送オペレーションが開始され、イベントリスナーが Pub/Sub キューからの通知を待機します。

転送オペレーションは、ステータスが in progress の状態で約 24 時間実行されます。24 時間後にオペレーションが終了し、新しいオペレーションが開始されます。転送ジョブの終了時刻に達するまで、またはジョブが手動で停止するまで、24 時間ごとに新しいオペレーションが作成されます。

オペレーションの終了がスケジュールされているときにファイル転送が進行中の場合、ファイルが完全に転送されるまで現在のオペレーションは進行中のままになります。新しいオペレーションが開始され、古いオペレーションが終了するまで 2 つのオペレーションが同時に実行されます。この期間中に検出されたイベントは、新しいオペレーションによって処理されます。

現在のオペレーションと完了したオペレーションを表示するには:

Google Cloud コンソール

コンソールの [Storage Transfer Service] ページに移動します。 Google Cloud

Storage Transfer Service に移動
ジョブリストで、[すべて] タブまたは [クラウド間] を選択します。
転送のジョブ ID をクリックします。[スケジュールモード] 列には、すべてのイベントドリブン転送とバッチ転送が示されます。
[オペレーション] タブを選択します。現在のオペレーションの詳細が表示され、完了したオペレーションが [実行履歴] テーブルに一覧表示されます。完了したオペレーションをクリックすると、詳細が表示されます。

gcloud

ジョブの進行状況をリアルタイムでモニタリングするには、gcloud transfer jobs monitor を使用します。レスポンスには、現在のオペレーション、ジョブの開始時間、転送されたデータの量、スキップされたバイト数、エラー数が表示されます。

gcloud transfer jobs monitor JOB_NAME

現在のオペレーション名を取得するには:

gcloud transfer jobs describe JOB_NAME --format="value(latestOperationName)"

現在と完了したオペレーションを一覧表示するには:

gcloud transfer operations list --job-names=JOB_NAME

オペレーションの詳細を表示するには:

gcloud transfer operations describe OPERATION_NAME

Cloud Storage からのイベント ドリブン転送 コレクションでコンテンツを整理 必要に応じて、コンテンツの保存と分類を行います。

権限を構成する

Cloud コンソール

gcloud CLI

Pub/Sub を構成する

転送ジョブを作成する

Cloud コンソール

gcloud

REST

クライアント ライブラリ

Go

Java

Node.js

Python

イベント ドリブン転送をモニタリングする

Google Cloud コンソール

gcloud

Cloud Storage からのイベントドリブン転送

`gcloud` CLI

クライアントライブラリ

イベントドリブン転送をモニタリングする