はじめに

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については、ここを確認ください。