概要

この記事の CosmosDB版です。指定した件数・間隔でデータを自動生成し、CosmosDBのデータベースに書き込む Python プログラムです。データの読み込み・削除も可能となっています。

ローカル環境

macOS Monterey 12.3.1
python 3.8.12
Azure CLI 2.34.1

前提条件

Azure環境がすでに用意されていること（テナント/サブスクリプション）
ローカル環境に「azure cli」がインストールされていること

事前準備

この記事を実行し、Azure CosmosDB をプロビジョニングされたスループット構成で作成できていること

項目	値
Endpoint	https://iturucosmosdb01.documents.azure.com:443/
CosmosDB アカウント名	iturucosmosdb01
データベース名	CPDemoDB01
コンテナ名	container01
パーティション	section

Pythonプログラムからの CosmosDB への操作

Pythonプログラム

cosmosdb_IoTdummy.py

import time
from datetime import date, datetime
import random
import json
import argparse
import string
from faker.factory import Factory
import pandas as pd
from azure.cosmos import exceptions, CosmosClient, PartitionKey

endpoint = 'https://iturucosmosdb01.documents.azure.com:443/'
key = 'xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx'
databaseName = 'CPDemoDB01'
containerName = 'container01'


# ダミーデータ作成のための Faker の使用
Faker = Factory.create
fake = Faker()
fake = Faker("ja_JP")

# ダミーセクション？(大文字アルファベットを定義)
section = string.ascii_uppercase


# JSONダミーデータの作成
def iot_json_data(count, proc):
    iot_items = json.dumps({
        'items': [{
            'id': str(datetime.now().timestamp()),  # id
            'ctid': i,                              # create id
            'proc': proc,                           # データ生成プロセス名
            'section': random.choice(section),      # IoT機器セクション
            'iot_num': fake.zipcode(),              # IoT機器番号
            'iot_state': fake.prefecture(),         # IoT設置場所
            'val_1': random.uniform(100, 200),      # IoT値−１
            'val_2': random.uniform(50, 90),        # IoT値−２
            'created_at': generate_time()           # データ生成時間
            } 
            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))


# CosmosDBへの接続
def ConnectionToCosmosDB():
    try:
      client = CosmosClient(endpoint, key)
      db = client.get_database_client(databaseName)
      container = db.get_container_client(containerName)
      return 1, container
    except Exception as err:
      return 0, err


# メイン : ターミナル出力用
def tm_main(count, proc, wait):
    print('ターミナル 出力\n')

    # ダミーデータ生成
    iotjsondata = iot_json_data(count, proc)
    json_dict = json.loads(iotjsondata)

    # ターミナル出力
    for item in json_dict['items']:
        print(item)
        time.sleep(wait)


# メイン : Show Database
def show_db_main():
    print('CosmosDB への接続確認')
    code, conn = ConnectionToCosmosDB()

    if code == 0 :
      print(conn)
    else :
      print(conn)


# メイン : Write Database
def write_db_main(count, proc, wait):
    print('データベースへのデータ書き込み')

    # DBへの接続
    code, conn = ConnectionToCosmosDB()

    if code == 0 :
      print(conn)
      return

    # ダミーデータ生成
    iotjsondata = iot_json_data(count, proc)
    json_dict = json.loads(iotjsondata)

    # データの書き込み
    try:
      for item in json_dict['items']:
        print(list(item.values()))
        conn.create_item(item)
        time.sleep(wait)
      print("\nInserted",count,"row(s) of data.")
      print("Done.")
    except Exception as err:
      print(err)


# メイン : Read Database
def read_db_main():
    print('データベースからのデータ読み込み')

    # DBへの接続
    code, conn = ConnectionToCosmosDB()

    if code == 0 :
      print(conn)
      return

    # Read table data
    items = conn.read_all_items()

    # print all rows
    try:
      [print(f'{item}') for item in items]
      print("Done.")
    except Exception as err:
      print(err)


# メイン : Delete Database
def delete_db_main():
    print('データベースのデータ全削除')

    # DBへの接続
    code, conn = ConnectionToCosmosDB()

    if code == 0 :
      print(conn)
      return

    try:
      for item in conn.query_items(query='SELECT * FROM container01',enable_cross_partition_query=True):
          print(item['section'])
          conn.delete_item(item, partition_key=item['section'])
      print("Done.")
    except Exception as err:
      print(err)


# メイン
if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='ダミーデータの自動生成からのDBへの書込み')
    parser.add_argument('--count', type=int, default=5, help='データ作成件数（デフォルト：5件）')
    parser.add_argument('--proc', type=str, default='111', help='データ作成プロセス名（デフォルト：111）')
    parser.add_argument('--mode', type=str, default='tm', help='tm（Terminal(デフォルト)）/ db（db操作）')
    parser.add_argument('--wait', type=float, default=1, help='データ生成間隔（デフォルト：1.0秒）')
    parser.add_argument('--db', type=str, default='show', help='show(デフォルト) / write / read / delete')
    args = parser.parse_args()

    start = time.time()

    if (args.mode == 'tm'): 
      tm_main(args.count, args.proc, args.wait)
    elif (args.mode == 'db'):
      if (args.db == 'show'):
        show_db_main()
      elif (args.db == 'write'):
        write_db_main(args.count, args.proc, args.wait)
      elif (args.db == 'read'):
        read_db_main()
      elif (args.db == 'delete'):
        delete_db_main()
    else :
      print("パラメータ設定を確認ください --help")

    making_time = time.time() - start

    print("")
    print("処理時間:{0}".format(making_time) + " [sec]")
    print("")

プログラムの実行

実行パラメータを指定することにより以下の処理が可能となります
- データ生成
  - パラメータ
    - --mode
      - tm : 生成データのターミナル出力（デフォルト）
      - db : DB操作の実施
    - --proc : 本プログラム実行場所の識別子（デフォルト：'111'）
    - --count : 生成データの件数（デフォルト：5件）
    - --wait : データ書込み時の間隔（秒）（デフォルト：1.0秒、最小：0秒）
- DB操作
  - パラメータ（上記「mode」パラメータで「db」を選択することにより有効となります）
    - --db
      - show : データベースへの接続確認（デフォルト）
      - write : テーブルへの生成データ書き込み
      - read : テーブルからの全件データ読み込み
      - delete : テーブルの全件データ削除

ヘルプ表示

$ python cosmosdb_IoTdummy.py -h                  
usage: cosmosdb_IoTdummy.py [-h] [--count COUNT] [--proc PROC] [--mode MODE] [--wait WAIT] [--db DB]

ダミーデータの自動生成からのDBへの書込み

optional arguments:
  -h, --help     show this help message and exit
  --count COUNT  データ作成件数（デフォルト：5件）
  --proc PROC    データ作成プロセス名（デフォルト：111）
  --mode MODE    tm（Terminal(デフォルト)）/ db（db操作）
  --wait WAIT    データ生成間隔（デフォルト：1.0秒）
  --db DB        show(デフォルト) / write / read / delete

ターミナルへの生成データの出力

$ python cosmosdb_IoTdummy.py   
ターミナル 出力

{'id': '1652363331.254743', 'ctid': 0, 'proc': '111', 'section': 'N', 'iot_num': '606-9267', 'iot_state': '京都府', 'val_1': 141.51992902165148, 'val_2': 67.06060158494354, 'created_at': '2022-05-12T22:48:51.254792'}
{'id': '1652363331.254798', 'ctid': 1, 'proc': '111', 'section': 'F', 'iot_num': '963-8665', 'iot_state': '山梨県', 'val_1': 190.62681621674932, 'val_2': 52.34280626979511, 'created_at': '2022-05-12T22:48:51.254835'}
{'id': '1652363331.254838', 'ctid': 2, 'proc': '111', 'section': 'Z', 'iot_num': '131-3410', 'iot_state': '熊本県', 'val_1': 199.5290412707653, 'val_2': 61.636601101310475, 'created_at': '2022-05-12T22:48:51.254867'}
{'id': '1652363331.254869', 'ctid': 3, 'proc': '111', 'section': 'H', 'iot_num': '041-2071', 'iot_state': '山形県', 'val_1': 144.47870335349302, 'val_2': 76.18492745985165, 'created_at': '2022-05-12T22:48:51.254881'}
{'id': '1652363331.254883', 'ctid': 4, 'proc': '111', 'section': 'H', 'iot_num': '883-0190', 'iot_state': '高知県', 'val_1': 198.43396241325718, 'val_2': 55.100609116813224, 'created_at': '2022-05-12T22:48:51.254894'}

処理時間:5.012450933456421 [sec]

DBへの接続確認

$ python cosmosdb_IoTdummy.py --mode db --db show
CosmosDB への接続確認
<ContainerProxy [dbs/CPDemoDB01/colls/container01]>

処理時間:0.23801898956298828 [sec]

データの書き込み

## ８件のデータを0.2秒間隔で書き込む
$ python cosmosdb_IoTdummy.py --mode db --db write --count 8 --wait 0.2
データベースへのデータ書き込み
['1652363480.937302', 0, '111', 'B', '640-6595', '長崎県', 124.29416912385994, 78.0063814862435, '2022-05-12T22:51:20.937444']
['1652363480.937457', 1, '111', 'O', '572-2593', '千葉県', 116.3744925587041, 76.85188192753809, '2022-05-12T22:51:20.937495']
['1652363480.937501', 2, '111', 'C', '854-2449', '石川県', 164.8104189841249, 70.86096920541297, '2022-05-12T22:51:20.937530']
['1652363480.937535', 3, '111', 'E', '326-5458', '新潟県', 177.18539247589842, 88.7860288485881, '2022-05-12T22:51:20.937578']
['1652363480.937583', 4, '111', 'T', '331-7151', '宮城県', 176.29096635579108, 58.605170597776066, '2022-05-12T22:51:20.937609']
['1652363480.937614', 5, '111', 'Q', '376-7413', '高知県', 114.1503599127303, 62.52303505813319, '2022-05-12T22:51:20.937639']
['1652363480.937644', 6, '111', 'C', '427-4899', '北海道', 185.81540242201626, 53.17876457695124, '2022-05-12T22:51:20.937669']
['1652363480.937673', 7, '111', 'K', '318-8102', '福井県', 139.14614950001638, 65.42312161402906, '2022-05-12T22:51:20.937698']

Inserted 8 row(s) of data.
Done.

処理時間:2.137359142303467 [sec]

データの全件読み込み

$ python cosmosdb_IoTdummy.py --mode db --db read                      
データベースからのデータ読み込み
{'id': '1652363480.937302', 'ctid': 0, 'proc': '111', 'section': 'B', 'iot_num': '640-6595', 'iot_state': '長崎県', 'val_1': 124.29416912385994, 'val_2': 78.0063814862435, 'created_at': '2022-05-12T22:51:20.937444', '_rid': 'PQFjAP5XJMoSAAAAAAAAAA==', '_self': 'dbs/PQFjAA==/colls/PQFjAP5XJMo=/docs/PQFjAP5XJMoSAAAAAAAAAA==/', '_etag': '"340014e3-0000-2300-0000-627d10db0000"', '_attachments': 'attachments/', '_ts': 1652363483}
{'id': '1652363480.937457', 'ctid': 1, 'proc': '111', 'section': 'O', 'iot_num': '572-2593', 'iot_state': '千葉県', 'val_1': 116.3744925587041, 'val_2': 76.85188192753809, 'created_at': '2022-05-12T22:51:20.937495', '_rid': 'PQFjAP5XJMoTAAAAAAAAAA==', '_self': 'dbs/PQFjAA==/colls/PQFjAP5XJMo=/docs/PQFjAP5XJMoTAAAAAAAAAA==/', '_etag': '"340015e3-0000-2300-0000-627d10db0000"', '_attachments': 'attachments/', '_ts': 1652363483}
{'id': '1652363480.937501', 'ctid': 2, 'proc': '111', 'section': 'C', 'iot_num': '854-2449', 'iot_state': '石川県', 'val_1': 164.8104189841249, 'val_2': 70.86096920541297, 'created_at': '2022-05-12T22:51:20.937530', '_rid': 'PQFjAP5XJMoUAAAAAAAAAA==', '_self': 'dbs/PQFjAA==/colls/PQFjAP5XJMo=/docs/PQFjAP5XJMoUAAAAAAAAAA==/', '_etag': '"340016e3-0000-2300-0000-627d10db0000"', '_attachments': 'attachments/', '_ts': 1652363483}
{'id': '1652363480.937535', 'ctid': 3, 'proc': '111', 'section': 'E', 'iot_num': '326-5458', 'iot_state': '新潟県', 'val_1': 177.18539247589842, 'val_2': 88.7860288485881, 'created_at': '2022-05-12T22:51:20.937578', '_rid': 'PQFjAP5XJMoVAAAAAAAAAA==', '_self': 'dbs/PQFjAA==/colls/PQFjAP5XJMo=/docs/PQFjAP5XJMoVAAAAAAAAAA==/', '_etag': '"340018e3-0000-2300-0000-627d10db0000"', '_attachments': 'attachments/', '_ts': 1652363483}
{'id': '1652363480.937583', 'ctid': 4, 'proc': '111', 'section': 'T', 'iot_num': '331-7151', 'iot_state': '宮城県', 'val_1': 176.29096635579108, 'val_2': 58.605170597776066, 'created_at': '2022-05-12T22:51:20.937609', '_rid': 'PQFjAP5XJMoWAAAAAAAAAA==', '_self': 'dbs/PQFjAA==/colls/PQFjAP5XJMo=/docs/PQFjAP5XJMoWAAAAAAAAAA==/', '_etag': '"34001be3-0000-2300-0000-627d10dc0000"', '_attachments': 'attachments/', '_ts': 1652363484}
{'id': '1652363480.937614', 'ctid': 5, 'proc': '111', 'section': 'Q', 'iot_num': '376-7413', 'iot_state': '高知県', 'val_1': 114.1503599127303, 'val_2': 62.52303505813319, 'created_at': '2022-05-12T22:51:20.937639', '_rid': 'PQFjAP5XJMoXAAAAAAAAAA==', '_self': 'dbs/PQFjAA==/colls/PQFjAP5XJMo=/docs/PQFjAP5XJMoXAAAAAAAAAA==/', '_etag': '"34001ce3-0000-2300-0000-627d10dc0000"', '_attachments': 'attachments/', '_ts': 1652363484}
{'id': '1652363480.937644', 'ctid': 6, 'proc': '111', 'section': 'C', 'iot_num': '427-4899', 'iot_state': '北海道', 'val_1': 185.81540242201626, 'val_2': 53.17876457695124, 'created_at': '2022-05-12T22:51:20.937669', '_rid': 'PQFjAP5XJMoYAAAAAAAAAA==', '_self': 'dbs/PQFjAA==/colls/PQFjAP5XJMo=/docs/PQFjAP5XJMoYAAAAAAAAAA==/', '_etag': '"34001de3-0000-2300-0000-627d10dc0000"', '_attachments': 'attachments/', '_ts': 1652363484}
{'id': '1652363480.937673', 'ctid': 7, 'proc': '111', 'section': 'K', 'iot_num': '318-8102', 'iot_state': '福井県', 'val_1': 139.14614950001638, 'val_2': 65.42312161402906, 'created_at': '2022-05-12T22:51:20.937698', '_rid': 'PQFjAP5XJMoZAAAAAAAAAA==', '_self': 'dbs/PQFjAA==/colls/PQFjAP5XJMo=/docs/PQFjAP5XJMoZAAAAAAAAAA==/', '_etag': '"34001ee3-0000-2300-0000-627d10dc0000"', '_attachments': 'attachments/', '_ts': 1652363484}
Done.

処理時間:0.23212695121765137 [sec]

データの全件削除

$ python cosmosdb_IoTdummy.py --mode db --db delete
データベースのデータ全削除
B
O
C
E
T
Q
C
K
Done.

処理時間:0.5690500736236572 [sec]

まとめ

これで、CosmosDBのデータ関連の各種テストがやりやすくなるかなぁ、、、、、

参考記事

以下の記事を参考にさせていただきました。感謝申し上げます。
Azure Cosmos DB SQL API client library for Python