Azure Blob StorageのデータをGrafanaとDuckDBで可視化してみた

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from azure.storage.blob import BlobServiceClient
import os
import argparse
from azure.core.exceptions import AzureError
import tempfile

def setup_azure_client(connection_string=None):
    """
    Azure Blob Storageクライアントを安全に設定
    """
    try:
        if not connection_string:
            connection_string = os.getenv('AZURE_STORAGE_CONNECTION_STRING')

        if not connection_string:
            raise ValueError("Azure Storage接続文字列が設定されていません")

        blob_service_client = BlobServiceClient.from_connection_string(connection_string)

        # 接続テスト
        account_info = blob_service_client.get_account_information()
        print(f"Azure Storage接続成功")
        print(f"アカウント種別: {account_info.get('account_kind', 'Unknown')}")

        return blob_service_client

    except AzureError as e:
        print(f"Azure接続エラー: {e}")
        raise
    except Exception as e:
        print(f"予期しないエラー: {e}")
        raise

def generate_iot_data(start_date, days=30):
    """IoTセンサーデータを生成"""
    data = []
    current_time = start_date

    for i in range(days * 24 * 6):  # 10分間隔でデータ取得
        # 温度データ（季節変動 + 日内変動 + ノイズ）
        temp_base = 20 + 10 * np.sin(2 * np.pi * i / (24 * 6))  # 日内変動
        temperature = temp_base + np.random.normal(0, 2)

        # 湿度データ（温度と逆相関 + ノイズ）
        humidity = 70 - (temperature - 20) * 2 + np.random.normal(0, 5)
        humidity = max(10, min(90, humidity))  # 10-90%の範囲

        data.append({
            'timestamp': current_time,
            'sensor_id': 'SENSOR_001',
            'temperature': round(temperature, 2),
            'humidity': round(humidity, 2),
            'location': 'Office_Room_A'
        })

        current_time += timedelta(minutes=10)

    return pd.DataFrame(data)

def upload_to_blob_storage(df, connection_string, container_name, blob_name):
    """Azure Blob Storageにデータをアップロード"""
    try:
        blob_service_client = setup_azure_client(connection_string)

        # コンテナ存在確認
        try:
            container_client = blob_service_client.get_container_client(container_name)
            container_properties = container_client.get_container_properties()
            print(f"コンテナ '{container_name}' への接続確認完了")
        except Exception as e:
            print(f"コンテナアクセスエラー: {e}")
            raise

        # 一時ファイルに保存してからアップロード
        with tempfile.NamedTemporaryFile(suffix='.parquet', delete=False) as temp_file:
            df.to_parquet(temp_file.name, index=False, engine='pyarrow')
            temp_file_path = temp_file.name

        # Blob Storageにアップロード
        blob_client = blob_service_client.get_blob_client(
            container=container_name, 
            blob=blob_name
        )

        with open(temp_file_path, 'rb') as data:
            blob_client.upload_blob(data, overwrite=True)

        print(f"アップロード完了: {container_name}/{blob_name}")

        # 一時ファイル削除
        os.unlink(temp_file_path)

    except Exception as e:
        print(f"Blob Storageアップロードエラー: {e}")
        raise

def main():
    # コマンドライン引数の設定
    parser = argparse.ArgumentParser(description='IoTセンサーデータをAzure Blob Storageにアップロード')
    parser.add_argument('--connection-string', '-c',
                       help='Azure Storage接続文字列')
    parser.add_argument('--container', '-n', default='iot-sensor-data',
                       help='Blob Storageコンテナ名 (デフォルト: iot-sensor-data)')
    parser.add_argument('--days', '-d', type=int, default=30,
                       help='生成するデータの日数 (デフォルト: 30日)')

    args = parser.parse_args()

    # 設定
    CONNECTION_STRING = args.connection_string or os.getenv('AZURE_STORAGE_CONNECTION_STRING')
    CONTAINER_NAME = args.container
    BLOB_NAME = 'sensor_data_2024_01.parquet'

    if not CONNECTION_STRING:
        print("エラー: Azure Storage接続文字列が必要です")
        print("--connection-string オプションまたは環境変数 AZURE_STORAGE_CONNECTION_STRING を設定してください")
        return

    print("=== IoTセンサーデータ生成・アップロード ===")
    print(f"コンテナ: {CONTAINER_NAME}")
    print(f"データ期間: {args.days}日")
    print()

    # データ生成
    start_date = datetime(2024, 1, 1)
    df = generate_iot_data(start_date, days=args.days)

    # タイムスタンプ型変換
    df['timestamp'] = pd.to_datetime(df['timestamp'])

    print(f"生成されたデータ: {len(df)} レコード")
    print(f"期間: {df['timestamp'].min()} ～ {df['timestamp'].max()}")
    print()

    # Azure Blob Storageにアップロード
    upload_to_blob_storage(df, CONNECTION_STRING, CONTAINER_NAME, BLOB_NAME)

if __name__ == "__main__":
    main()