はじめに

S3のストレージクラスを適切に設定することはコストの最適化に繋がります。
ある程度分析可能な過去データがある場合であれば、ストレージクラス分析によってライフサイクルを決定することができますが、新規構築の場面などではそれが難しく、一度運用してから調整する流れになるかと思います。

今回は事前にある程度リクエスト数を想定して、その際にどのようなS3配置を取ればいいのかシミュレーションする方法を検討してみました！

新しい技術としてAmazon Braketを使用しており、Braketの使用感を確認したい方、定式化の方法を確認したい方にもお楽しみいただける記事です。

やりたいこと

以下のようなケースを考えてみたいと思います。
総計1000GB分のオブジェクトがあり、それらをS3 Standard、S3 Standard-IA、S3 Glacier Instant Retrievalに分類する場面を考えます。

S3のコストは大きく分けると保管コストとリクエストに分けて考えることができます。これらは以下の表に示している通り、トレードオフの関係を持っています。したがって、S3のコストの最適化を目指す場合には、保管量とアクセス量どちらも考慮したうえでストレージクラスを検討していく必要があります。

今回は各アクセス数（変数）に対して、どの配置がコスト最適であるのかを検討していきたいともいます。

組み合わせ最適化

今回S3の最適な配置を考えていくためのアプローチとして、「組み合わせ最適化」を使用していこうと思います。

組み合わせ最適化とは、様々な制約の中で、多くの取りうる選択肢の中から価値を最大化する組み合わせを求めることです。
代表的な組み合わせ最適化問題としてナップサック問題が広く知られています。

今回S3の配置を考えるために、以下のように12個の要素の組み合わせ最適化問題として考えていきます。

今回取りうるストレージクラスとしてはStandard、IA、Glacier Instant Retrievalの3種類、容量に関しては100, 200, 400, 800としています。容量は前述の4種類を用いることで合計1000になるような組み合わせを再現することができます。
例えばStandard:300GB, IA:500GB, Gracier:200GBを表現するには、

Standard : 100 + 200
IA : 100 + 400
Glacier : 200

の組み合わせを取ればよいということになります。すなわち、12個の要素の最適な組み合わせを求めることで、S3の配置の最適化が可能になるということになります。

Amazon Braket

組み合わせ最適化を行うにあたり、今回はAmazon Braketを使ってみようと思いました。（あまり事例も多くはなく、いつか使ってみたかった想いもあり...）

Amazon Braketとはフルマネージド量子コンピューティングサービスです。

量子コンピュータは大きく分けると2つの計算方式があります。
Braketではかつては"D-wave"という、量子アニーリング方式のQPUを使用することが可能でしたが、現在ではマーケットプレイス経由での購入のみであり、オンデマンドで使用することはできなくなりました。

そのため今回は量子ゲート方式で問題を解いていこうと思います。

注意
インターネット上には過去にbraketからD-waveを使った量子ゲート方式での検証記事が散見されますが、現在は使えないです。

QAOAアルゴリズム

QAOA (Quantum Approximation Optimization Algorithm) とは量子ゲート方式で組み合わせ最適化を行うためのアルゴリズムです。
詳しいアルゴリズムについては別記事がわかりやすいので以下をご参照ください。

今回はQUBO(Quadratic Unconstrained Binary Optimization)形式で定式化を行い、QAOAアルゴリズムで最適化を行っていきます。
QUBOでは以下に示すように最適化対象と制約の和が最小となるように式を組み立てていきます。

H (ハミルトニアン) = コスト(最小化対象) + 制約

定式化

変数設定/制約

変数
今回組み合わせ最適化の要に対して上記のようにバイナリ変数(0 or 1)の$x_i$を設定していきます。また横軸を表現するために、$size_i$を定義しています。$mod$は合同式を表しており、$i$を4で割った時のあまりで条件を分けています。

制約
今回の最適化では、組み合わせ要素の各容量を全て足し合わせると1000GBになることが必要であり、これが制約になります。

QUBO形式でこれを表現するためには、"制約の最小化 = 最適"としたいので、全容量が1000に近い時に最小の値を取る2次関数として$Restriction$を設定しました。

コスト

保管コスト
保管コストは前述のコスト表に沿って決定されます。
今、変数設定として、Standard: $1\leq i\leq 4$、IA: $5\leq i\leq 8$、Glacier: $9\leq i\leq 12$としているため、それぞれで条件分岐したコストを設定します。

保管データのGB単位での課金になるため、各ストレージクラスのコストと各組み合わせ要素のサイズの積を求め、$StoredCost$としました。

リクエストコスト
リクエストコストも同様に前述のコスト表に沿って決定されます。

リクエスト数単位での課金になるため、GET/PUTリクエストコストと各組み合わせ要素のリクエスト数の積を求め、$RequestCost$としました。
各要素のリクエスト数はどのオブジェクトに対しても均一にリクエストが行われることを前提としているため、$size_i / 1000$のようにデータ保管量の割合で導出しています。

ハミルトニアン設定

前節までで求めたコスト項と制約項の和をハミルトニアンとします。
今回制約項の係数として$k=10^{-2}$としました。
(本来であれば強い制約をかけるため、係数を大きくしますが、今回コスト項の値は制約項の値と比べると大きく桁が異なります。そのため、係数によって桁数の調整ならびに制約の強さを補正しています。)

実装

項目	設定
プログラム開発環境	マネージドJupyter Notebook
言語	python
主要ライブラリ	Qiskit (量子コンピュータ用フレームワーク) \n pyqubo (QUBO形式定式化ツール)
実行デバイス	SV1(ゲート方式のマネージドシミュレータ)

実装コード

# QiskitはマネージドJupyter Notebookにデフォルトでインストールされていますが、以下文献の通り再インストールが必要です。
# https://zenn.dev/yuhino/articles/8d3006b6396e76
%pip uninstall -y qiskit
%pip install --upgrade qiskit
%pip install --upgrade qiskit-optimization
%pip install --upgrade qiskit-algorithms
%pip install --upgrade pyqubo
%pip install --upgrade numpy

import numpy as np
from pyqubo import Binary
from qiskit_algorithms.utils import algorithm_globals
from qiskit_optimization import QuadraticProgram
from qiskit_algorithms import QAOA, NumPyMinimumEigensolver
from qiskit_algorithms.optimizers import COBYLA
from qiskit_optimization.algorithms import MinimumEigenOptimizer
from qiskit.primitives import BackendSampler
from qiskit_braket_provider import BraketProvider
from braket.tracking import Tracker
from qiskit_braket_provider import AWSBraketProvider

# 最大サイズ
max_data_size = 1000

# 変数設定
binary_set = [Binary("x1"), Binary("x2"), Binary("x3"), Binary("x4"), Binary("x5"), Binary("x6"), Binary("x7"), Binary("x8"), Binary("x9"), Binary("x10"), Binary("x11"), Binary("x12")]
data_size = [100, 200, 400, 800]

# コスト設定
stored_cost_dict = {"Standard": 0.025, "IA": 0.0138, "Glacier": 0.005}
get_request_cost_dict = {"Standard": 0.00000037, "IA": 0.000001, "Glacier": 0.00001}
put_request_cost_dict = {"Standard": 0.0000047, "IA": 0.00001, "Glacier": 0.00002}

# 変数
get_request = 100000
put_request = 100000

# 最小化対象Hc, 制約Hwの計算
Hc = 0
Hw = max_data_size
for i in range(12):
    k = data_size[ i % 4 ]
    x = binary_set[ i ]

    if i < 4:
        stored_cost = stored_cost_dict["Standard"]
        get_request_cost = get_request_cost_dict["Standard"]
        put_request_cost = put_request_cost_dict["Standard"]
    elif i < 8:
        stored_cost = stored_cost_dict["IA"]
        get_request_cost = get_request_cost_dict["IA"]
        put_request_cost = put_request_cost_dict["IA"]
    else:
        stored_cost = stored_cost_dict["Glacier"]
        get_request_cost = get_request_cost_dict["Glacier"]
        put_request_cost = put_request_cost_dict["Glacier"]
    Hc += k * x * (stored_cost + (get_request_cost * get_request + put_request_cost * put_request) / max_data_size)
    Hw -= (k * x)

# ハミルトニアンの計算
k = 10**-3
H = Hc + k * Hw**2

# QUBO式へ変換
model = H.compile()
print(model.variables)
quadratic, offset = model.to_qubo()
print(quadratic, offset)

# qbit設定
qubo = QuadraticProgram()
qubo.binary_var("x1")
qubo.binary_var("x2")
qubo.binary_var("x3")
qubo.binary_var("x4")
qubo.binary_var("x5")
qubo.binary_var("x6")
qubo.binary_var("x7")
qubo.binary_var("x8")
qubo.binary_var("x9")
qubo.binary_var("x10")
qubo.binary_var("x11")
qubo.binary_var("x12")

# デバイス設定
provider = AWSBraketProvider()
sampler = BackendSampler(
    backend=provider.get_backend("SV1"),
    options={"shots": 1000},
)

qubo.minimize(quadratic=quadratic)
algorithm_globals.random_seed = 10598
qaoa_mes = QAOA(sampler=sampler, optimizer=COBYLA(), initial_point=[0.0, 0.0])
qaoa = MinimumEigenOptimizer(qaoa_mes)  # using QAOA
# exact_mes = NumPyMinimumEigensolver()
# exact = MinimumEigenOptimizer(exact_mes)  # using the exact classical numpy minimum eigen solver

# 実行
with Tracker() as tracker:
    result = qaoa.solve(qubo)
    # result = exact.solve(qubo)
    
print("process Completed")

# 結果整理
def cal_cost(stored_cost_dict, get_request_cost_dict, put_request_cost_dict, get_request, put_request, max_data_size, data_size_result_dict):
    result_cost = {}
    sum = 0
    for storage_class in data_size_result_dict.keys():
        stored_cost = data_size_result_dict[storage_class] * stored_cost_dict[storage_class]
        get_request_cost = get_request_cost_dict[storage_class] * get_request * data_size_result_dict[storage_class] / max_data_size
        put_request_cost = put_request_cost_dict[storage_class] * put_request * data_size_result_dict[storage_class] / max_data_size
        sub_total = stored_cost + get_request_cost + put_request_cost
        result_cost[storage_class] = {"stored_cost": stored_cost, "get_request_cost": get_request_cost, "put_request_cost": put_request_cost, "sub_total": sub_total}
        sum += sub_total
    
    result_cost["total"] = sum
        
    return result_cost

# 実行条件表示
print(qubo.prettyprint())
print("Get Requests : " + str(get_request) + ", Put Requests : " + str(put_request))
print("k : " + str(k))
print("------------------------------------------------------------------------------------")

# 結果表示
for sample in result.samples:
    subtotal = 0
    sum = []

    for i, data in enumerate(sample.x):
      subtotal += data_size[i % 4] * data

      if i % 4 == 3:
        sum.append(int(subtotal))
        subtotal = 0
    
    # set result dict for calculate cost
    data_size_result_dict = {"Standard": sum[0], "IA": sum[1], "Glacier": sum[2]}
    
    result_cost = cal_cost(stored_cost_dict, get_request_cost_dict, put_request_cost_dict, get_request, put_request, max_data_size, data_size_result_dict)
    
    # print result
    total_size = sum[0] + sum[1] + sum[2]
    
    if total_size == 1000:
        print("Result : " + str(sample))
        print("-> Standard : " + str(sum[0]) + " GB, IA : " + str(sum[1]) + " GB, Glacier : " + str(sum[2]) + " GB.")
        print("Sum : " + str(sum[0] + sum[1] + sum[2]) + "GB.")
        print("Cost : " + str(result_cost))
        print("------------------------------------------------------------------------------------")

# print braket result
print(f"Task statistics: {tracker.quantum_tasks_statistics()}")
print(f"Estimated cost to run this example: {tracker.qpu_tasks_cost():.3f} USD")

結果

braketの計算結果は以下のようになりました。

量子計算が正しい答えを導くことができるのかを、一般的なソルバーで答えを求めて確認してみました。今回の検証では一般的なソルバーの方が良い結果をもたらすことがわかりました。(量子コンピュータで最適解が求められなかったことは悔しいですが...)

原因分析
以下のような原因があると考えています。

量子コンピュータは確率論として計算が行われるため、、値の振れ幅が小さいと最適を見誤る可能性がある
→ コストに基づいていたため数値が非常に小さかった
試行回数が少なかった/計算量が十分でなかった
→ 最適解(確度の高い解)にたどり着く前に終了してしまった
量子コンピュータのチューニングが十分にできていない。
→ 量子コンピュータを扱う上での知見を十分に蓄える必要がある。

おわりに

今回検証としては最適解は求められなかったものの、Amazon braketを扱うための定式化や実行方法を示すことはできました。今後量子コンピュータの活用が広がることを期待しています。

おまけ

量子コンピュータを使う方法

コードの中で以下の部分を変更すると、他のコードはそのままで、バックエンドで任意の量子コンピュータを実行することができます。

:
sampler = BackendSampler(
-   backend=provider.get_backend("SV1"),
+   backend=provider.get_backend("Aria 1"),
    options={"shots": 1000},
)
:

サポートされている量子デバイスは以下を参照ください。

課金体系

Braketの料金体系は"タスク数"と"shot数"に依存します。
1タスクの中で多数のShotを行うことが多々あると思うので、QPUを使用するときには料金を常に意識したほうがよいです。

またタスクは非同期的に実行されるため、コード上止めていても、裏で実行され続けることがあります。

参考