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 SQS に送信された Amazon S3 イベント通知をリッスンして、ソースバケット内のオブジェクトが変更または追加されたタイミングを確認します。オブジェクトの削除は検出されません。転送元でオブジェクトを削除しても、転送先バケット内の関連オブジェクトは削除されません。

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

始める前に

手順に沿って、宛先 Cloud Storage バケットに必要な権限を付与します。

宛先へのアクセスを構成する: Cloud Storage

SQS キューを作成する

AWS コンソールで [Simple Queue Service] ページに移動します。
[Create queue] をクリックします。
このキューの名前を入力します。

[Access policy] セクションで、[Advanced] を選択します。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 を置き換えます。
[Create queue] をクリックします。

完了したら、キューの Amazon Resource Name（ARN）をメモします。

S3 バケットで通知を有効にする

AWS コンソールで [S3] ページに移動します。
[Buckets] リストで、ソースバケットを選択します。
[Properties] タブをクリックします。
[Event notifications] で [Create event notification] をクリックします。
このイベントの名前を指定します。
[Event types] で、[All object create events] を選択します。
[Destination] で [SQS queue] を選択し、この転送用に作成したキューを選択します。
[変更を保存] をクリックします。

権限を構成する

ソースへのアクセスを構成する: Amazon S3 の手順で、アクセスキー ID と秘密鍵を作成するか、フェデレーション ID ロールを作成します。

手順に沿って操作する際は、カスタムロールまたはカスタム信頼ポリシーを指定するよう指示されたら、次の 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 と秘密鍵をメモします。
フェデレーション ID ロールの場合は、arn:aws:iam::AWS_ACCOUNT:role/AWS_ROLE_NAME という形式の Amazon Resource Name（ARN）をメモします。

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 と秘密鍵をメモします。
フェデレーション ID ロールの場合は、arn:aws-us-gov:iam::AWS_ACCOUNT:role/AWS_ROLE_NAME という形式の Amazon Resource Name（ARN）をメモします。

転送ジョブを作成する

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

Cloud コンソール

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

[転送ジョブを作成] に移動
ソースタイプとして Amazon S3、宛先として Cloud Storage を選択します。
[スケジュールモード] として [イベントドリブン] を選択し、[次のステップ] をクリックします。
S3 のバケット名を入力します。このバケット名は、AWS Management Console に表示される名前です。例: 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 キューで通知を待機します。ジョブの詳細ページには、1 時間ごとに 1 つのオペレーションが表示されます。このページには、各ジョブで転送されたデータの詳細も表示されます。

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 は、作成したフェデレーション ID ロールの 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 のリファレンスドキュメントをご覧ください。


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


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


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 からのイベントドリブン転送

クライアントライブラリ