※本記事は、Mayank Kakani による“Uploading whole directories to Oracle Object Storage”を翻訳したものです。
Mayank Kakani
Cloud Architect
2022年 3月 10日
パブリック・クラウドへの移行は、簡単なタスクではありません。ときには、数十年分に値するデータを、アプリケーション・ログやプライベート・ドキュメントの形でクラウド・オブジェクト・ストレージへ移行する必要があります。多くの場合、オブジェクト・ストレージへ移行して保守性と管理性を高めると同時に、オンプレミス環境にあるのと同じディレクトリ構造を維持することが求められます。
オブジェクト・ストレージとは
Oracle Cloud Infrastructure（OCI）Object Storageサービスは、インターネット規模の高パフォーマンスのストレージ・プラットフォームで、信頼性が高くコスト効率に優れたデータ永続性を提供します。Object Storageサービスでは、分析データや、画像やビデオなどのリッチ・コンテンツを含む、あらゆるコンテンツ・タイプの非構造化データを無制限に保存できます。
従来の統合のユースケースでは、ファイル・サーバーやSFTPアクセスに大きく頼ってファイルを保管しています。これらに急速に置き換わりつつあるのが、Object Storageや同様のサービスです。ますます多くのお客様が、これらのサービスのメリットを利用して、保存や移動を必要とするあらゆる種類のコンテンツを処理しています。
このブログでは、ファイルのアップロードにPython SDKを利用するシンプルなPythonスクリプトを用いて、元のディレクトリ構造を維持しながらオンプレミスのディレクトリをOCI Object Storageへアップロードする方法をご紹介します。
OCI CLIを設定する
該当ファイルがあるローカル・サーバーにOCIコマンドライン・インタフェース（CLI）を設定します。OCI CLIのインストールと構成に関するガイドの手順に従ってください。CLIを構成したら、ターミナルでoci os ns getコマンドを実行してこれをテストできます。
バケットを作成する
OCI Object Storageでバケットを作成します。ここに、すべてのファイルとディレクトリをアップロードします。バケットを作成するには、Oracle Object Storageバケットの作成に関するガイドに従ってください。私は、ディレクトリのアップロード先として、bucket-directory-uploadという名前のバケットを作成しました。
コンパートメントIDとバケット名を取得する
次に、スクリプトが使用するバケットに関する情報が必要となります。必要なのは、バケットが作成されるコンパートメントIDと、バケットの名前です。OCIコンパートメントになじみがなく、これについて詳しく知りたいという方は、Oracle Cloud Infrastructure Compartmentsを参照してください。作成したバケットからコンパートメントIDを取得できます。
Bucket Informationタブに、関連するコンパートメントが表示されます。これをクリックすると、コンパートメントの詳細ページが表示されます。このページのコンパートメントの詳細タブで、コンパートメントのOCIDを取得できます。
スクリプトを記述する
これから記述するスクリプトには、入力としてアップロードするルート・ディレクトリが必要です。スクリプトにはPython SDKを使用します。自宅で作業を行っていて、詳しい情報を知りたい場合は、Python SDKへのリンクがあります。
このスクリプトを実行するには、次の前提条件が必要です。
- 必要なアクセス権があるルート・ディレクトリ
- Pythonの基礎知識（筆者はPython 3.6を使用しています）
- Pythonでのプロセスと同時実行性
以下の図は、スクリプトが機能する様子を示したものです。
では、コードに移りましょう。このコードでは、4つの関数のワークフローをレプリケートします。
processDirectory
この関数は、ディレクトリに対して反復され、現在のアイテムがディレクトリの場合はプロセス・ディレクトリを呼び出し、現在のアイテムがファイルの場合はprocessDirectoryObjectsを呼び出します。
"""
processDirectoryは、現在のディレクトリを処理します
:param path:現在のディレクトリのパス
:param object_storage_client:オブジェクト・ストレージのSDKクライアント
:param namespace:バケットの名前空間
:proc_list:クライアント・リスト。クライアントは、個々のプロセスが同じクライアントを参照しないように呼出しのたびに作成されます。
"""
def processDirectory(path:Path,object_storage_client,namespace,proc_list):
if path.exists():
print("in directory ---- " + path.relative_to(p).as_posix())
for objects in path.iterdir():
if objects.is_dir():
processDirectory(objects,object_storage_client,namespace,proc_list)
else:
processDirectoryObjects(object,object_storage_client,namespace,proc_list)
processDirectoryObjects
この関数は、現在のパスを含むアイテムがファイルかどうかをチェックします。ファイルである場合、createUploadProcess関数が呼び出されます。
"""
processDirectoryObjectsは、現在のオブジェクト・パスがファイルかそうでないかをチェックし、ファイルの場合はアップロード・プロセスを作成します
:param object: アップロードするオブジェクトのパス
:param object_storage_client: オブジェクト・ストレージのSDKクライアント
:param namespace: バケットの名前空間
:param proc_list: クライアント・リスト。クライアントは、個々のプロセスが同じクライアントを参照しないように呼出しのたびに作成されます。
"""
def processDirectoryObjects(object:Path,object_storage_client,namespace,proc_list):
if object.is_file():
createUploadProcess(object,object_storage_client,namespace,proc_list)
createUploadProcess
この関数は、ルート・ディレクトリに関連するアイテムの完全な名前を取得し、同時実行性のためのセマフォを取得してプロセスを作成します。プロセスの作成後、プロセスをリストに追加して開始します。
"""
createUploadProcessは、同時アップロード・プロセスを作成し、それをproc_listに入れます。
:param object: アップロードするオブジェクトのパス
:param object_storage_client: オブジェクト・ストレージのSDKクライアント
:param namespace: バケットの名前空間
:param proc_list:クライアント・リスト。クライアントは、個々のプロセスが同じクライアントを参照しないように呼出しのたびに作成されます。
"""
def createUploadProcess(object:Path,object_storage_client,namespace,proc_list):
name = object.relative_to(p).as_posix()
sema.acquire()
process = Process(target=upload_to_object_storage, args= (object.as_posix(),name,object_storage_client,namespace))
proc_list.append(process)
process.start()
upload_to_object_storage
この関数は、ファイルを読み取り、それをObject Storageへアップロードします。アップロード後、セマフォを解放します。
"""
upload_to_object_storageは、オブジェクト・ストレージ・バケットへファイルをアップロードします。この関数は、独立したプロセスとして実行されます。クライアントは、個々のプロセスが同じクライアントを参照しないように呼出しのたびに作成されます。
:param path:アップロードするオブジェクトのパス
:param name:アップロードするオブジェクトの名前
:param object_storage_client:オブジェクト・ストレージのSDKクライアント
:param namespace:バケットの名前空間
"""
def upload_to_object_storage(path:str,name:str,object_storage_client,namespace):
with open(path, "rb") as in_file:
object_storage_client.put_object(namespace,bucket_name,name,in_file)
print("Finished uploading {}".format(name))
sema.release()すべてのコードをメイン・ブロックと一緒にuploadToOSS.pyというファイルにまとめましょう。
"""
from array import array
from pathlib import Path
import oci
from multiprocessing import Process
from multiprocessing import Semaphore
# 1回に可能なプロセスの最大数
concurrency= 5
sema = Semaphore(concurrency)
# ルート・ディレクトリ・パス、ユーザーのパスで置換え
p = Path('/Users/mkakani_in/Documents/blockchain')
# コンパートメントOCID
compartment_id =
"ocid1.compartment.oc1..aaaaaaaa2hacvkp2z726xmpv3ykqjox4anmmzrq2d2ok2je53p7g4dojktwa"
# アップロードするバケットの名前
bucket_name = "bucket-directory-upload"
"""
upload_to_object_storageは、オブジェクト・ストレージ・バケットへファイルをアップロードします。この関数は、独立したプロセスとして実行されます。クライアントは、個々のプロセスが同じクライアントを参照しないように呼出しのたびに作成されます。
:param path:アップロードするオブジェクトのパス
:param name:アップロードするオブジェクトの名前
:param object_storage_client:オブジェクト・ストレージのSDKクライアント
:param namespace:バケットの名前空間
"""
def upload_to_object_storage(path:str,name:str,object_storage_client,namespace):
with open(path, "rb") as in_file:
print("Starting upload {}", format(name))
object_storage_client.put_object(namespace,bucket_name,name,in_file)
print("Finished uploading {}".format(name))
sema.release()
"""
createUploadProcessは、同時アップロード・プロセスを作成し、それをproc_listに入れます。
:param object:アップロードするオブジェクトのパス
:param object_storage_client:オブジェクト・ストレージのSDKクライアント
:param namespace:バケットの名前空間
:param proc_list:クライアント・リスト。クライアントは、個々のプロセスが同じクライアントを参照しないように呼出しのたびに作成されます。
"""
def createUploadProcess(object:Path,object_storage_client,namespace,proc_list):
name = object.relative_to(p).as_posix()
sema.acquire()
process = Process(target=upload_to_object_storage, args=(object.as_posix(),name,object_storage_client,namespace))
proc_list.append(process)
process.start()
"""
processDirectoryObjectsは、現在のオブジェクト・パスがファイルかそうでないかをチェックし、ファイルの場合はアップロード・プロセスを作成します
:param object:アップロードするオブジェクトのパス
:param object_storage_client:オブジェクト・ストレージのSDKクライアント
:param namespace:バケットの名前空間
:param proc_list:クライアント・リスト。クライアントは、個々のプロセスが同じクライアントを参照しないように呼出しのたびに作成されます。
"""
def processDirectoryObjects(object:Path,object_storage_client,namespace,proc_list):
if object.is_file():
createUploadProcess(object,object_storage_client,namespace,proc_list)
"""
processDirectoryは、現在のディレクトリを処理します
:param path:現在のディレクトリのパス
:param object_storage_client:オブジェクト・ストレージのSDKクライアント
:param namespace:バケットの名前空間
:proc_list:クライアント・リスト。クライアントは、個々のプロセスが同じクライアントを参照しないように呼出しのたびに作成されます。
"""
def processDirectory(path:Path,object_storage_client,namespace,proc_list):
if path.exists():
print("in directory ---- " + path.relative_to(p).as_posix())
for objects in path.iterdir():
if objects.is_dir():
processDirectory(objects,object_storage_client,namespace,proc_list)
else:
processDirectoryObjects(objects,object_storage_client,namespace,proc_list)
if __name__ == '__main__':
config = oci.config.from_file()
object_storage_client = oci.object_storage.ObjectStorageClient(config)
namespace = object_storage_client.get_namespace().data
proc_list: array = []
sema = Semaphore(concurrency)
if p.exists() and p.is_dir():
processDirectory(p,object_storage_client,namespace,proc_list)
for job in proc_list:
job.join() ここでは、マルチプロセッシング用のプロセスとセマフォを使用しています。これによって、最大5つのファイルをOracle Object Storageに同時にアップロードできます。ここでは同時実行性を5に定義しましたが、変更が可能です。
さて、コマンドpython3 uploadToOSS.pyを用いてスクリプトを実行する準備ができました。
スクリプトはうまくいきました。コンソールを通じてObject Storageバケットをチェックし、ファイルがアップロードされているかどうかを見てみましょう。
筆者のファイルはすべて、ローカル・マシンにあったのと同じディレクトリ構造でOCI Object Storageにアップロードされているようです。
まとめ
このブログでは、Object Storageについて、そしてシンプルなPythonスクリプトを用いて、元のオンプレミスのディレクトリ構造を維持しながらオンプレミスのデータをOCI Object Storageへアップロードする方法についてご紹介しました。
Oracle Cloud Infrastructureでは、開発者が最新のクラウド・アプリケーションを構築するためのエンタープライズ機能を使用できます。このブログの内容を無償でお試しになりたい場合は、300米ドル分のクレジットが30日間使用できる無償トライアルが提供されるOracle Cloud Free Tierをお薦めします。Free Tierには、無償クレジットの期限が切れてからも無期限に使用できるAlways Freeサービスもいくつか含まれます。