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
)