STSを用いずに、S3からGCSへのファイル転送を実現した話

はじめに

Google Cloud Platform（GCP）のStorage Transfer Service（STS）をご存知でしょうか？これはGoogleが提供する、異なるストレージシステム間でのデータ移行を自動化するためのサービスです。特に、大量のデータをGoogle Cloud Storage（GCS）に移行したい場合や、定期的なバックアップが必要な場合に適しています。STSは、単発の転送を行う「Batch」転送と、AWSなどの他のクラウドサービスのイベントを検知して転送を実行する「Event Driven」転送の二つのモードを提供しています。コードを記述することなく使用できるサービスなので、私のようなエンジニア初心者（今年で１年）でも扱いやすいサービスです。

一方で、STSはすべてのケースに最適なわけではありません。例えば、転送したいファイルが複数存在し、S3のバケット、もしくはディレクトリも多岐にわたる場合、その分だけ対応したSTSを作成する必要があります。「Event Driven」で作成する場合、リソースの管理が難しくなることもあると思います。

今回は、上記のSTSを使わずに、S3とGCS間でファイル転送を行う手順をご紹介します。
STSの代替手段として、AWS Lambdaを使用しております。
メリットとしましては、ログをAWSに集約できること、複雑なディレクトリ構造にも柔軟に対応できるということが挙げられます。
（参考）
Storage Transfer Service とは

手順

① サービスアカウントの作成

GCPの「IAMと管理」より、「サービスアカウント」を選択します。
「サービスアカウントの作成」を選択し、1.「サービス アカウントの詳細」を設定します。（２、３は省略してよいです）

② サービスアカウントキーの作成とダウンロード

①で作成したサービスアカウントを一覧より選択し、「キー」タブを選択します。「鍵の追加」より「新しい鍵を作成」を選択します。
鍵のタイプで「JSON」を選択し、「作成」をクリック。ファイルをダウンロードします。
ダウンロードしたファイルはわかりやすいように、test.jsonのように変更しておきます。
（参考）
サービス アカウント キーの作成と管理

③ Google Cloud Storageへのアクセス権の付与

Google Cloud Storage（GCS）へ遷移します。転送したいGCSのバケットを選択して、権限タブを選択します。
その後、アクセス権を付与を選択します。
「新しいプリンシパル」より、先ほど作成したサービスアカウントを選択します。
「ロールを選択」より、「Storage レガシー オブジェクト オーナー」「ストレージ管理者」を選択し、作成したサービスアカウントに必要な権限を付与します。

④ Lambdaレイヤーの作成

今回はLambda関数を作成し、S3、GCS間の転送を行います。
AWSのサービス一覧から、Lambdaを選択。「レイヤー」より、「google.cloud」パッケージをインストールしたレイヤーを作成します。
（参考）
【初心者向け】AWS Lambda Layerの作成方法をわかりやすく解説【Python3.9】

⑤ Lambda関数の作成

Lambda関数の作成より、以下のLambdaを作成します。（Pythonで作成しています）
その際、②でダウンロードしたサービスアカウントのキー（test.json）をLambda内に配置します。
実行の際は、④で作成したレイヤーをアタッチしておきます。
（転送したいデータの大きさにより、メモリ等の設定を変更する必要があります）

import json,os
import boto3
from botocore.exceptions import ClientError
from google.cloud import storage
from google.oauth2.service_account import Credentials

# GCS転送用の関数
def upload_to_gcs(credentials, gcs_project_id, bucket_name, source_file_name, destination_blob_name, content_type):
    storage_client = storage.Client(credentials=credentials, project=gcs_project_id)
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(destination_blob_name)

    blob.upload_from_string(source_file_name, content_type=content_type)
    

def lambda_handler(event, context):
    s3 = boto3.client('s3')
    
    # 転送先のGCSバケット
    gcs_project = "your-gcp-project" 
    gcs_bucket = "your-gcs-bucket"
    gcs_directory = "your-gcs-directory"
    
    with open('test.json') as f:
        credentials_info = json.load(f)
    credentials_test = Credentials.from_service_account_info(credentials_info)
    
    try:
        # S3コピー元
        src_bucket = "your-s3-bucket"
        src_prefix = "your-s3-prefix"
        
        # コピー元のファイル一覧を取得
        src_files = s3.list_objects(Bucket=src_bucket, Prefix=src_prefix)
        
        for file in src_files['Contents']:
            # 各ファイルを取得
            src_key = file['Key']
            dest_key = src_key.replace(src_prefix, gcs_directory, 1)
            obj = s3.get_object(Bucket=src_bucket, Key=src_key)['Body'].read()
            
            # GCSへ転送
            put_file = os.path.basename(dest_key)
            upload_to_gcs(credentials_test, gcs_project, gcs_bucket, obj, gcs_directory + put_file, 'text/csv')
            
    except ClientError as e:
        error_code = e.response["Error"]["Code"]
        print(f"file cannot be downloaded. Error code: {error_code}")

⑥ Lambda関数のテスト実行

「テスト」よりLambda関数を実行します。
GCSを確認すると、ファイルが正しく転送されていることがわかります！

あとはLambdaでS3トリガーなど設定すれば、任意のタイミングでファイル転送ができそうです。

おわりに

今回の記事では、STSの代わりにLambda関数を利用してS3とGCS間でファイルを転送する方法をご紹介しました。
STSも扱いやすいサービスなので、目的に応じて使い分けていきたいですね。

今回の記事が同様の問題に直面している方のお役に立てば幸いです。