Streamlitアプリのパフォーマンス改善と運用戦略

import streamlit as st
import squadbase.streamlit as sq

# ユーザー情報の取得
user_info = sq.auth.get_user()
st.write(f"ようこそ、{user_info['firstName']}さん")

# 役割に基づく表示制御
if "admin" in user_info['roles']:
    st.header("管理者向けダッシュボード")
elif "analyst" in user_info['roles']:
    st.header("アナリスト向けダッシュボード")
else:
    st.info("このページを表示する権限がありません。")

@st.fragment
def real_time_metrics():
    """リアルタイム指標の表示"""
    col1, col2, col3 = st.columns(3)
    with col1:
        st.metric("売上", get_daily_sales())
    # 自動更新処理
    time.sleep(5)
    st.rerun()

import streamlit as st
import pandas as pd

@st.cache_data(ttl=3600)  # 1時間キャッシュ
def load_sales_data():
    """売上データの読み込み（データベースから）"""
    # 重いデータベース処理
    query = "SELECT * FROM sales WHERE date >= '2024-01-01'"
    return pd.read_sql(query, connection)

@st.cache_data(ttl=1800)  # 30分キャッシュ
def get_aggregated_metrics():
    """集計済み指標の取得"""
    df = load_sales_data()
    return {
        'total_sales': df['amount'].sum(),
        'avg_order': df['amount'].mean(),
        'customer_count': df['customer_id'].nunique()
    }

# 使用例
sales_df = load_sales_data()
metrics = get_aggregated_metrics()

st.dataframe(sales_df)
st.metric("総売上", f"¥{metrics['total_sales']:,.0f}")

@st.cache_resource
def init_database_connection():
    """データベース接続の初期化（アプリ起動時に1回のみ）"""
    return create_engine(DATABASE_URL)

@st.cache_resource
def load_ml_model():
    """機械学習モデルの読み込み（メモリに常駐）"""
    return joblib.load('sales_prediction_model.pkl')

# 使用例
conn = init_database_connection()
model = load_ml_model()

# リアルタイム性が重要なデータ（5分キャッシュ）
@st.cache_data(ttl=300)
def get_current_metrics():
    return fetch_real_time_data()

# 日次更新データ（6時間キャッシュ）
@st.cache_data(ttl=21600)
def get_daily_report():
    return generate_daily_summary()

# マスターデータ（24時間キャッシュ）
@st.cache_data(ttl=86400)
def get_product_master():
    return fetch_product_catalog()

@st.cache_data
def get_filtered_data(date_range, category, user_id):
    """ユーザーごと、条件ごとに個別キャッシュ"""
    # パラメータが異なれば自動的に別のキャッシュとして保存
    return filter_data(date_range, category, user_id)

# 使用例：異なる条件で個別にキャッシュされる
data_admin = get_filtered_data("2024-01", "electronics", "admin")
data_user1 = get_filtered_data("2024-01", "books", "user1")

# レベル1: 生データのキャッシュ
@st.cache_data(ttl=3600)
def load_raw_sales_data():
    return pd.read_sql("SELECT * FROM sales", connection)

# レベル2: 加工データのキャッシュ
@st.cache_data(ttl=1800)
def get_processed_data():
    raw_data = load_raw_sales_data()
    # データクレンジングと基本集計
    processed = raw_data.groupby(['date', 'category']).agg({
        'amount': ['sum', 'mean', 'count']
    }).reset_index()
    return processed

# レベル3: 表示用データのキャッシュ
@st.cache_data(ttl=900)
def get_dashboard_summary():
    processed_data = get_processed_data()
    # 最終的なダッシュボード用サマリー
    return create_summary_table(processed_data)

# 問題1: キャッシュが効かない
# 原因: 関数の引数が可変オブジェクト（リスト、辞書など）
# 解決策: hash_funcs を使用

@st.cache_data(hash_funcs={list: lambda x: str(sorted(x))})
def process_categories(category_list):
    return analyze_categories(category_list)

# 問題2: メモリ不足
# 原因: 大容量データの長時間キャッシュ
# 解決策: max_entries でキャッシュ数を制限

@st.cache_data(ttl=3600, max_entries=10)
def load_large_dataset(dataset_name):
    return load_massive_data(dataset_name)

# 問題3: 古いデータの表示
# 原因: TTLが長すぎる
# 解決策: 手動キャッシュクリアの実装

if st.button("データを最新化"):
    st.cache_data.clear()
    st.rerun()

@st.cache_data
def analyze_large_dataset(file_path):
    df = pl.scan_csv(file_path)
    return (df
            .filter(pl.col("amount") > 1000)
            .groupby("category") 
            .agg(pl.col("amount").sum())
            .collect())

# 良い例：データベースで集計
@st.cache_data(ttl=1800)
def get_sales_summary():
    query = "SELECT category, SUM(amount) FROM sales GROUP BY category"
    return pl.read_database(query, connection)

-- 簡略化されたテーブル構造例
CREATE TABLE sales_fact (
    sale_id INT PRIMARY KEY,
    date_key INT,
    product_key INT, 
    sales_amount DECIMAL
);

この設計により、複雑な分析クエリを高速実行できます。

```python
# データマート更新の基本パターン
def update_sales_mart():
    """売上データマートの日次更新"""
    # 1. 複雑な集計を事前計算
    query = """
    SELECT date, category, SUM(amount) as total_sales
    FROM sales 
    WHERE date >= current_date - 90
    GROUP BY date, category
    """
    
    # 2. 結果をマートテーブルに保存
    save_to_mart(query)
    
    print("データマート更新完了")

# Streamlitアプリでは軽量クエリのみ
@st.cache_data(ttl=1800)
def get_dashboard_data():
    return fetch_from_mart("SELECT * FROM sales_mart")