はじめに
AWS SDK for Python(boto3) を使って、プログラムをほぼ変更せずに Cloudian と S3 へのアクセスが可能となります。オブジェクトストレージをハイブリッド（オンプレミス：Cloudian、AWS：S3）で使いたい方への参考になればと、、、
概要
オブジェクトストレージ Cloudian/S3 上のバケット名「boto3-cloudian」に、JSON形式のデータを書込む Python プログラムです。
生成するJSON形式のデータ件数はパラメータで指定でき、ファイル「test-iot-dummy.json」に書き込まれます。プログラムをカスタマイズすることにより、なんちゃってIoTデータ生成として使えるかもと想定しております。生成されるデータ項目についてはプログラム内の「items」を参照ください。
- パラメータは以下の３種類となります。
- --count ： 生成するデータ件数（デフォルト：10件）
- --proc ： 生成するプロセス名（デフォルト：111）
- --mode ： 生成データの出力先の指定 tm：ターミナルへの出力、 s3：Cloudian/S3への出力（デフォルト：tm）
プログラム実行時に、パラメータ「-h」を指定することにより表示されるヘルプも参照ください。
実行環境
macOS Big Sur 11.1
python 3.8.3
クレデンシャル情報の定義
今回はクレデンシャル情報を .zshenv に定義してプログラムを実行しています。接続先に合わせて定義ください。
# AWS S3
export AWS_ACCESS_KEY_ID=xxxxxxxxxxxxx
export AWS_SECRET_ACCESS_KEY=yyyyyyyyyyyyyyyyy
export AWS_DEFAULT_REGION=ap-northeast-1
# Cloudian
#export AWS_ACCESS_KEY_ID=aaaaaaaaaaaaaaaaaa
#export AWS_SECRET_ACCESS_KEY=bbbbbbbbbbbbbbbbbbbb
#export AWS_DEFAULT_REGION=pic
実行プログラム
Cloudianへアクセスする場合は endpoint_url を記載ください（プログラム内を参照ください）。
IoTSample-write.py
import random
import json
import time
from datetime import date, datetime
from collections import OrderedDict
import argparse
import string
import boto3
import pprint
from faker.factory import Factory
BUCKET_NAME = 'boto3-cloudian'
OBJECT_KEY = 'test-iot-dummy.json'
# ダミーデータ作成のための Faker の使用
Faker = Factory.create
fake = Faker()
fake = Faker("ja_JP")
# IoT機器のダミーセクション(小文字アルファベットを定義)
section = string.ascii_uppercase
# IoT機器で送信JSONデータの作成
def iot_json_data(count, proc):
iot_items = json.dumps({
'items': [{
'id': i, # id
'time': generate_time(), # データ生成時間
'proc': proc, # データ生成プロセス名
'section': random.choice(section), # IoT機器セクション
'iot_num': fake.zipcode(), # IoT機器番号
'iot_state': fake.prefecture(), # IoT設置場所
'vol_1': random.uniform(100, 200), # IoT値−１
'vol_2': random.uniform(50, 90) # IoT値−２
}
for i in range(count)
]
}, ensure_ascii=False).encode('utf-8')
return iot_items
# IoT機器で計測されたダミーデータの生成時間
def generate_time():
dt_time = datetime.now()
gtime = json_trans_date(dt_time)
return gtime
# date, datetimeの変換関数
def json_trans_date(obj):
# 日付型を文字列に変換
if isinstance(obj, (datetime, date)):
return obj.isoformat()
# 上記以外は対象外.
raise TypeError ("Type %s not serializable" % type(obj))
# メイン（ターミナル出力用）
def tm_main(count, proc):
print('ターミナル 出力')
iotjsondata = iot_json_data(count, proc)
pprint.pprint(iotjsondata)
# メイン（Cloudian/S3 出力用）
def s3_main(count, proc):
print('Cloudian/S3 出力')
iotjsondata = iot_json_data(count, proc)
# pprint.pprint(iotjsondata)
# client = boto3.client('s3', endpoint_url='http://s3-pic.networld.local') # Cloudianへのアクセス時
client = boto3.client('s3') # S3へのアクセス時
client.put_object(
Bucket=BUCKET_NAME,
Key=OBJECT_KEY,
Body=iotjsondata
)
if __name__ == '__main__':
parser = argparse.ArgumentParser(description='IoT機器のなんちゃってダミーデータの生成')
parser.add_argument('--count', type=int, default=10, help='データ作成件数')
parser.add_argument('--proc', type=str, default='111', help='データ作成プロセス名')
parser.add_argument('--mode', type=str, default='tm', help='tm（生成データをターミナル出力）／ s3（生成データをCloudian/S3出力）')
args = parser.parse_args()
start = time.time()
if (args.mode == 's3'):
s3_main(args.count, args.proc)
else :
tm_main(args.count, args.proc)
making_time = time.time() - start
print("")
print(f"データ作成件数:{args.count}")
print("データ作成時間(通常_API):{0}".format(making_time) + " [sec]")
print("")
プログラムの実行
最初にヘルプを表示してみます。
$ python IoTSample-write.py -h
usage: IoTSample-write.py [-h] [--count COUNT] [--proc PROC] [--mode MODE]
IoT機器のなんちゃってダミーデータの生成
optional arguments:
-h, --help show this help message and exit
--count COUNT データ作成件数
--proc PROC データ作成プロセス名
--mode MODE tm（生成データをターミナル出力）／ s3（生成データをCloudian/S3出力）
次に、10万件のデータを生成し、それをターミナル出力してみます
$ python IoTSample-write.py --count 100000
：
出力内容は割愛
：
データ作成件数:100000
データ作成時間(通常_API):4.5370988845825195 [sec]
今度は実際に Cloudian/S3 に10万件のデータを生成してみます。
$ python IoTSample-write.py --count 100000 --mode s3
データ作成件数:100000
データ作成時間(通常_API):2.7221038341522217 [sec]
まとめ
今回は、AWS SDK for Python(boto3) を使って、オブジェクトストレージ Cloudian / S3 へデータを生成することを確認できました（10万件のデータを数秒で作成完了（もちろん環境に依存します））。
Cloudianについては、ここ を確認ください。