PythonでAWS S3のフォルダ配下の最新ファイルを取得 & 履歴管理する方法

ML環境で一定間隔（日次とか週次）で学習モデルが更新されていく運用の場合、AWSであれば学習モデルや学習データをS3で管理する場合が多いと思います。

本記事では以下の状況を想定し、boto3でPythonからS3を操作する方法を説明します。

• S3に置かれている最新の学習データや学習モデルを常に入力として用いる
• 学習データや学習モデルは一定ファイル数のみを履歴保持する

環境

• Mac Mojave 10.14.3
• Python 3.7.3

Sessionオブジェクト作成

ます、以下によりboto3.session.Sessionオブジェクトを作成します。

from boto3.session import Session

session = Session(
    aws_access_key_id = '************',
    aws_secret_access_key = '************',
    region_name = '************'
)

S3の特定ディレクトリ配下の最新ファイルをダウンロード

以下のメソッドにより、特定ディレクトリ配下の最新ファイルをダウンロードできます。

def download_recent_file(
    session: Session,
    s3_bucket: str,
    s3_prefix: str,
    dist_file_path: str
) -> None:
    '''
    S3から特定のフォルダ配下の最新ファイルをダウンロードするメソッド。

    Parameters
    ----------
    session: Session
        boto3.sessionオブジェクト。
    s3_bucket: string
        最新ファイルをダウンロードしたいS3バケット名。
    s3_prefix: string
        最新ファイルをダウンロードしたいS3フォルダパス。
    dist_file_path: string
        ダウンロード先のファイルパス。
    '''

    s3 = session.resource("s3")
    bucket = s3.Bucket(s3_bucket)

    # S3から学習済みモデルをダウンロード
    # 学習済みモデルが格納されているS3ディレクトリ内のファイルリストオブジェクトを作成
    objs = bucket.meta.client.list_objects_v2(
        Bucket=bucket.name,
        Prefix=s3_prefix
    )
    # ディレクトリ配下のファイルについてLOOP処理
    # 最新の更新日付のファイルがダウンロードされる
    loop_first_f = True
    for o in objs.get('Contents'):
        # LOOP初回処理
        if loop_first_f:
            download_target_file = o.get('Key')
            modified_datetime_mid = o.get('LastModified')
            loop_first_f = False
        # 2回目以降
        else:
            # 最新の修正日時のファイル名を保持
            if modified_datetime_mid <= o.get('LastModified'):
                modified_datetime_mid = o.get('LastModified')
                download_target_file = o.get('Key')

    try:
        # S3からファイルをダウンロード
        bucket.download_file(
            download_target_file,
            dist_file_path
        )
        print(f'download {download_target_file} is completed.')

    except Exception:
        print(f'download {download_target_file} is failed.')

S3の特定ディレクトリ配下において履歴管理

以下のメソッドにより、特定ディレクトリ配下で履歴管理できます。
（一定ファイル数を超えた場合、古いファイルのみをパージしているだけです）

def delete_old_files(
    session: Session,
    s3_bucket: str,
    delete_target_dir: str,
    retention_file_cnt: int
) -> None:
    '''
    S3のフォルダ配下で履歴保持数以上のファイルがある場合、古いファイルを削除するメソッド。

    Parameters
    ----------
    session: Session
        boto3.sessionオブジェクト。
    s3_bucket: string
        対象のS3バケット名。
    delete_target_dir: string
        削除対象のS3ディレクトリ。
    retention_file_cnt: integer
        履歴保持したいファイル数。
    '''

    s3 = session.resource("s3")
    bucket = s3.Bucket(s3_bucket)
    # S3フォルダ内のファイルリストオブジェクトを作成
    objs = bucket.meta.client.list_objects_v2(
        Bucket=bucket.name,
        Prefix=delete_target_dir
    )

    s3_client = session.client('s3')

    if len(objs.get('Contents')) > retention_file_cnt:
        # S3に格納されているオブジェクトを格納する
        # {更新日: key_path}のdict
        object_dict = {}
        # 対象ディレクトリ配下のオブジェクトについてLOOP処理、辞書に格納
        for o in objs.get('Contents'):
            object_dict[o.get('LastModified')] = o.get('Key')

        delete_index_upbound = len(objs.get('Contents')) - retention_file_cnt
        for key in sorted(list(object_dict.keys()))[:delete_index_upbound]:
            s3_client.delete_object(Bucket=s3_prefix, Key=object_dict[key])
            print(f'{s3_prefix}{key} is deleted.')

実装例

上記で定義したメソッドを使って、以下のように実装すればOKです。

# 変数定義
s3_bucket = 'S3バケット名'
s3_prefix = '対象のS3フォルダ' # hoge/fuga/
dist_file_path = 'ダウンロード先のローカルファイルパス' # ./hoge.txt
delete_target_dir = '削除対象のフォルダ(履歴管理したいフォルダ)'
retention_file_cnt = 10 # 履歴保持したいファイル数

# S3から最新ファイルをダウンロード
download_recent_file(
	session,
	s3_bucket,
	s3_prefix,
	dist_file_path
)
# 履歴管理
delete_old_files(
	session,
	s3_bucket,
	delete_target_dir,
	retention_file_cnt
)