Snowflake

Streamlit in Snowflakeにおける分離コンテナ環境とセッション管理の仕組みを理解した話

投稿日
カテゴリ: Snowflakeタグ:

はじめに

Streamlit in Snowflakeで本番環境のアプリケーションを構築する際、実行環境とセッション管理の仕組みを理解することは必須である。標準的なStreamlitとは異なり、Snowflake統合版はSnowflakeの管理するコンテナ内で実行され、アプリケーションのライフサイクル、パフォーマンス特性、状態管理が大きく異なる。本稿では、この実行モデルの核心部分に焦点を当て、本番環境での実装判断に必要な知識を整理する。標準的なStreamlitの開発経験がある技術者であっても、Snowflake統合版の独特なアーキテクチャを把握することで、より堅牢で効率的なアプリケーション設計が可能となる。

Snowflakeの管理するコンテナ内での実行

Streamlit in Snowflakeのアプリケーションは、Snowflakeのアカウント内で管理された隔離されたコンテナプロセス上で実行される。ローカルマシンのPythonプロセスのように直接制御することはなく、Snowflakeのインフラストラクチャが実行環境全体を統制する。

実行環境の核心的な特性：

  • 各アプリケーションはSnowflakeのアカウント領域内で独立した仮想環境として分離されており、他のテナントや他のアプリケーションとの干渉を受けない
  • アプリケーションの起動、実行、終了はSnowflakeの制御下にあり、ユーザーのアクセスパターンに応じた動的なスケーリングが自動的に実行される
  • Pythonランタイムは事前にコンテナ内にプリロードされており、ユーザーがアプリケーションにアクセスした時点でコードの実行が即座に開始される
  • コンテナはステートレスな設計であり、複数のユーザーセッション間でローカルのファイルシステム上の状態は保持されない
  • メモリ、CPU、ネットワーク帯域幅などのリソースは制限されており、無限に大規模なデータセットをメモリに展開することはできない

この設計により、スケーラビリティと管理負荷の削減が実現される。開発者はインフラストラクチャの保守運用から解放され、アプリケーション本体の開発に集中できる。一方で、アプリケーション開発者は「各セッションは独立している」「ローカル状態は永続しない」という前提でコーディングする必要があり、この認識がなければ本番環境で予期しない動作が発生する可能性がある。

ExecutionContextとSnowflakeのセッション情報へのアクセス

Streamlit in Snowflakeで最も重要な概念がExecutionContextである。これはSnowflakeのセッション情報とアプリケーション実行の状態を統合したオブジェクトであり、アプリケーションコード内から直接アクセスすることが可能である。

ExecutionContextを通じて、認証済みユーザーの識別子、割り当てられたウェアハウス、セッションのロール情報、現在のデータベースとスキーマといった情報が取得できる。これらの情報はSnowflakeの権限管理体系と一体化しており、アプリケーションが実行するすべてのSQL文はこのコンテキストの権限に基づいて検証される。


from snowflake.snowpark.context import get_active_session

session = get_active_session()

# 現在のユーザーを取得
current_user = session.sql("SELECT CURRENT_USER()").collect()[0][0]

# 割り当てられたウェアハウスを確認
current_warehouse = session.sql("SELECT CURRENT_WAREHOUSE()").collect()[0][0]

# 現在のロール情報
current_role = session.sql("SELECT CURRENT_ROLE()").collect()[0][0]

# アプリケーション領域のスキーマを取得
current_schema = session.sql("SELECT CURRENT_SCHEMA()").collect()[0][0]

ExecutionContextから取得可能な情報の実用的な用途：

  • 認証済みユーザーID：このユーザーが属するテナント、部門、権限レベルをデータベースから検索し、表示内容を動的に制御する基盤となる
  • 割り当てられたウェアハウス：クエリの実行リソースがどのウェアハウスに割り当てられるかを把握し、多くの重い処理が実行される時間帯を避けるといった最適化判断に活用される
  • セッションのロール情報：ロールベースアクセス制御の実装において、現在のユーザーが実行可能な操作を制限する際に利用される
  • 現在のデータベースおよびスキーマ：アプリケーションが参照するテーブルやストアドプロシージャの名前空間を把握し、正確なクエリを構築する際に用いられる

ExecutionContextはSnowflakeの行レベルセキュリティ（RLS）および動的データマスキング（DDM）と組み合わせることで、マルチテナント環境でのデータ分離を実装できる。ユーザーが属するテナント情報をExecutionContextから抽出し、その情報をSQLクエリに動的にフィルタリング条件として付与するパターンが一般的である。

セッション状態の管理と永続化戦略

Streamlit in Snowflakeでは、標準的なStreamlitの`st.session_state`メカニズムが使用される。ただし、その永続性と可用性については、通常のStreamlitと異なる考慮が必要である。

セッション状態の保持期間と動作：

  • ユーザーがブラウザを閉じるまで、またはセッションのタイムアウト（デフォルト約60分）が発生するまで、`st.session_state`に格納されたPythonオブジェクトは保持される
  • セッション終了後、メモリ上の状態は完全に消失し、その後のユーザーアクセスでは初期化された状態から再出発する
  • 複数のユーザーセッションが並行して実行される場合、各セッションのメモリ空間は完全に独立しており、相互干渉は発生しない
  • 分散環境ではコンテナのリバランシングが発生する可能性があり、メモリ内状態への依存度が高いと予期しない状態喪失が発生する危険性がある

セッション状態を効果的に使用するパターンとしては、ユーザーの入力フォーム状態、フィルタ条件、ページネーション状態、一時的なキャッシュなど、セッション内での短期的な状態に限定することが推奨される。


import streamlit as st
from snowflake.snowpark.context import get_active_session

session = get_active_session()

# セッション状態で一時的なUIフィルタを保持
if 'selected_date_range' not in st.session_state:
    st.session_state.selected_date_range = (None, None)

if 'filter_status' not in st.session_state:
    st.session_state.filter_status = 'all'

# ユーザーインタラクションでセッション状態を更新
date_range = st.date_input("期間を選択", st.session_state.selected_date_range)
st.session_state.selected_date_range = date_range

# 永続化が必要な設定はSnowflakeテーブルに明示的に保存
if st.button('設定を保存'):
    current_user = session.sql("SELECT CURRENT_USER()").collect()[0][0]
    session.sql(f"""
        UPDATE user_preferences 
        SET ui_settings = parse_json(?)
        WHERE user_id = ?
    """, params=[str(st.session_state.filter_status), current_user]).collect()
    st.success("設定を保存しました")

セッション終了後も保持する必要があるデータ（ユーザー設定、保存された検索条件、レポート結果など）は、Snowflakeのテーブルに明示的に書き込む必要がある。この明確な分離により、アプリケーションの動作が予測可能になり、バグの温床となる隠れた状態共有が回避される。

パフォーマンス特性とコールドスタート最適化

Streamlit in Snowflakeのパフォーマンス特性は、コンテナの起動時間、リソースの割り当て、クエリの実行効率によって大きく影響を受ける。

パフォーマンスに関わる重要な指標：

  • 初期化時間：ユーザーがアプリケーションに初めてアクセスする際、Snowflakeがコンテナを起動し、Pythonランタイムを初期化するまでに3秒から10秒程度要する場合がある。これをコールドスタートと呼ぶ
  • SQLクエリ実行時間：SQLクエリの実行時間は主にSnowflakeのクエリプランニングと分散処理の時間に依存し、ネットワークレイテンシは相対的に最小限である
  • メモリ制限：各コンテナプロセスのメモリは制限されており、gigabyte単位の大規模なデータセットを一度にメモリにロードすることは技術的に不可能である
  • リソース競合：同一のウェアハウス上で複数のアプリケーションやクエリが並行実行される場合、リソース争奪による性能低下が発生する可能性がある
  • キャッシュ効果：頻繁にアクセスされるテーブルやクエリ結果はSnowflakeの内部キャッシュに保持され、2度目以降のアクセスは高速化される

本番環境ではコールドスタート対策として、アプリケーション初期化時の処理を最小化し、必要なデータは遅延読み込みするパターンが採用される。また、複雑な分析処理やデータ変換はSnowflakeのストアドプロシージャに委譲し、アプリケーション層では結果の表示と対話的なUIの提供に専念することが効率的である。

ウェアハウスとリソース割り当ての考慮

Streamlit in Snowflakeのアプリケーションが実行するすべてのSQL文は、指定されたウェアハウスのコンピュート能力を消費する。ウェアハウスの選択は、クエリの実行速度、同時実行可能なセッション数、運用コストに大きな影響を与える。

ウェアハウス選択の実務的考慮：

  • 小規模なウェアハウス（XSMALL、SMALL）はコストが低く、軽量なクエリや低アクセス頻度のアプリケーションに適しており、一方で大量のユーザーからの並行アクセスには不向きである
  • 大規模なウェアハウス（LARGE、XLARGE以上）は並行クエリ処理の能力が高く、多くのユーザーからの同時アクセスに対応できるが、アイドル状態であってもコストが発生する
  • オートスケール機能を有効にすることで、負荷に応じたウェアハウスの自動拡張が可能になり、ピーク時の対応と非ピーク時のコスト削減を両立させられる
  • 複数のアプリケーションが同一ウェアハウスを共有する場合、負荷分散戦略を立案し、一つのアプリケーションの過度なリソース消費が他のアプリケーションに悪影響を与えないようにする必要がある

リソースの効率的な利用と高いパフォーマンスの両立は、ウェアハウスのサイズ選択、クエリの最適化、適切なキャッシング戦略によって初めて実現される。

まとめ

Streamlit in Snowflakeは、Snowflakeの管理するマネージドコンテナ環境内で動作し、ExecutionContextを通じてSnowflake側のセッション情報に直接アクセスできるアーキテクチャである。セッション状態は短期的なUI状態の保持に限定し、永続化が必要なデータはSnowflakeテーブルに委譲することが設計の原則である。また、コールドスタートやリソース共有の課題を念頭に置いて、初期化処理の最小化とクエリの最適化によるパフォーマンス改善アプローチを検討する必要がある。これらの理解があれば、本番環境での実装判断が格段に容易になり、堅牢で拡張性の高いアプリケーション設計が可能となる。

関連記事

default eye-catch image.

幾何分布と「過去の結果からは何もわからない話」

[mathjax] いつか起こる大地震がもし昨日起きたとしたら明日起こる確率は下がるのか？ 飛行機が昨日落ちたとしたらしばらくは飛行機は落ちないのか？ うまくいかない人生が今日またうまくいかなかったとして将来うまくいかない確率は下がるのか？ 起こるか起こらないかの確率が変わらないのであれば、将来は過去に影響されないらしい。 影響されるかどうか、と聞かれるとされない、と答えるだろうけど、それを説明することができるらしい。 そんな無記憶性の件読んでみたのでまとめてみる。 統計学入門とこちらを参考にさせて頂きました。 幾何分布 幾何分布 二項分布、ポアソン分布は(n)回の試行のうち(x)回事象(A)が発生したときを話題にしていた。 前提を少し変えて、予め試行する回数を決めないで、 (x)回目の試行で初めて事象(A)が発生した、 という話をすることもできる様子。 確率変数(x)が等間隔に並ぶ時刻であるとすることで、 事象(A)が発生するまでの待ち時間に関する確率分布を作れる。 (x)回目の試行で初めて事象(A)が起こった、ということを、 (x-1)回事象(bar{A})が起こり、次に事象(A)が起きたと考える。 事象(A)の生起確率を(p)、事象(bar{A})の生起確率を(q)とすると、 その確率は以下のようになる。 begin{eqnarray} f(x) = p cdot q^{x-1} end{eqnarray} (x)の増加1回に対して(q)を1回かける構造で、公比(q)の等比数列になっている。 等比数列って英語でgeometric seriesって言うもんだから、幾何分布っていう名前がついてる様子。 (f(x))の確率変数(x)は事象(A)が発生するまでの試行回数（時間）である、と読む様子。 では確率変数(X)が幾何分布に従うとき、期待値はどうかというと(frac{1}{p})となる。 確率の以下の読み方から、(x)回の試行に平均して(frac{1}{p})回かかる、というのは妥当に見える。 事象(A)の生起確率(p)の読み方について、その逆数(frac{1}{p})は、事象(A)が起こるまでの試行回数と読む。 (p)回の試行で初めて事象(A)が起こった、というシーンで事象(A)の生起確率を(p)と置いているので..。 期待値 幾何分布の期待値は(frac{1}{p})。ベルヌーイ試行の確率が(p)であるならば、平均(frac{1}{p})回で事象が起こる。 分散は(frac{1-p}{p^2})。期待値の証明は以下の通り。へぇ。 begin{eqnarray} E(X) &=& sum_x x cdot f(x) \\ &=& sum_x x cdot p q^{x-1} end{eqnarray} (E(X)がfrac{1}{p})であることを示したい。 恒等式を使ってやる奴ではなく、愚直にやる奴を書く。 まず、(q=1-p)として(E(X))を変形しておいてスタート。 begin{eqnarray} E(X) &=& sum_x x cdot p (1-p)^{x-1} \\ &=& p sum_x x cdot (1-p)^{x-1} end{eqnarray} 右辺を生み出すために、(frac{1}{x-1})のテイラー展開を持ち出す。 begin{eqnarray} frac{1}{1-x} &=& 1 + x + x^2 + cdots \\ &=& sum_{k=0}^{infty} x^k end{eqnarray} 左辺を(x)で微分すると以下の通り。 begin{eqnarray} left( frac{1}{1-x} right) frac{d}{dx} = frac{1}{(1-x)^2} end{eqnarray} 右辺を(x)で微分すると以下の通り。 begin{eqnarray} sum_{k=0}^{infty} x^k frac{d}{dx} = sum_{k=1}^{infty} k x^{k-1} end{eqnarray} なので、 begin{eqnarray} frac{1}{(1-x)^2} = sum_{k=1}^{infty} k x^{k-1} end{eqnarray} (x=1-p)として式変形すると、 begin{eqnarray} frac{1}{p^2} = sum_{k=1}^{infty} k (1-p)^{k-1} end{eqnarray} (E(X))にこれらを代入すると、 begin{eqnarray} E(X) &=& p sum_x x cdot (1-p)^{x-1} \\ &=& p cdot frac{1}{p^2} \\ &=& frac{1}{p} end{eqnarray} 本当に(frac{1}{p})になった。 両辺を微分したものが等しいって、なんでだっけ？ 無記憶性 どうも、世の中には(n-1)回連続して失敗して(n)回目で初めて成功することを言っているものと、 (n)回連続して失敗して、(n+1)回目で初めて成功することを言っているものがある。 期待値も分散も若干違うものになる。 ここからは(n)回の失敗に続いて(n+1)回目で初めて成功するケースに切り替える。 その時の確率を(P(X=n))とする。で、失敗が(n)回以上連続して起こる確率(P(Xgeq n))を考える。 begin{eqnarray} P(X geq n) &=& P(X=n) + P(X=n+1) + P(X=n+2) + cdots \\ &=& p(1-p)^n + p(1-p)^{n+1} + p(1-p)^{n+2} + cdots \\ &=& p(1-p) left( 1 + (1-p)^1 + (1-p)^2 + cdots right) \\ &=& p(1-p) sum_{k=1}^{infty} (1-p)^{k-1} end{eqnarray} 途中の無限級数は(frac{1}{1-x})の級数展開になっていて、 以下みたいになる。 begin{eqnarray} P(X geq n) &=& p(1-p)^n sum_{k=1}^{infty} (1-p)^{k-1} \\ &=& p(1-p)^n frac{1}{1-(1-p)} \\ &=& p(1-p)^n frac{1}{p} \\ &=& (1-p)^n end{eqnarray} (n)回連続して失敗した上で、さらに連続して(k)回の失敗を重ねる確率を考える。 (n)回連続して失敗する確率は(P(Xgeq n))。 この条件の上でさらに(k)回失敗を重ねる確率は条件付き確率として(P(Xgeq n+k | X geq n))。 条件付き確率の定義と乗法定理から式を展開していく。（ここが難しかった...) begin{eqnarray} P(Xgeq n+k | X geq n) &=& frac{P((X geq n+k)cap (X geq n) )}{P(X geq n)} \\ &=& frac{P(X geq n+k)}{P(X geq n)} \\ &=& frac{(1-p)^{n+k}}{(1-p)^n} \\ &=& (1-p)^k \\ &=& P(X=k) end{eqnarray} ということで、以下が成り立つことがわかる。 begin{eqnarray} P(Xgeq n+k | X geq n) &=& P(X=k) end{eqnarray} よーく見てみると、(n)回連続して失敗した後に(k)回連続して失敗する確率と、 (n)回の失敗無しに、最初から(k)回連続して失敗する確率が同じである、と言っている。 凄まじいことに、(n)回連続して失敗することは、次の(k)回の失敗に全く影響を及ぼさない、と言っている。 何回失敗しようと次に失敗する確率はこれまでの失敗に影響されない。 つまり失敗する確率は過去の影響を受けない。 美しすぎる感じがする。

default eye-catch image.

線形サポートベクトル分類器で画像認識するテスト

線形サポートベクトル分類器で画像認識する流れを理解したので、 定着させるために記事にしてみます。 当然、モデルの数学的な理解がないとモデルを解釈することは不可能だし、 正しいハイパーパラメータを設定することも不可能なので、数学的な理解は不可欠。 NumPy、pandas、matplotlibに慣れないと、そこまで行くのに時間がかかります。 こちらはPythonプログラミングの領域なので、数こなして慣れる他ないです。 機械学習用のサンプル画像で有名なMNISTを使ってNumPy、pandasの練習。 手書き文字認識用の画像データを読み込んでみる。サイズは28x28。各々1byte。 MNISTの手書き文字認識画像の読み込み まず読み込んでみて、データの形を出力してみる。 X_trainは、要素が3個のTupleが返る。3次。 1番外が60000。28x28の2次のndarrayが60000個入っていると読む。 1枚目の画像データはX_train[0]によりアクセスできる。 import tensorflow as tf minst = tf.keras.datasets.mnist (X_train,y_train),(X_test,y_test) = mnist.load_data() print(X_train.shape) # (60000, 28, 28) y_trainは要素が1個のTupleが返る。1次。 1枚目から60000枚目までの画像が0から9のいずれに分類されたかが入っている。 y_train[0]が4なら、1枚目の画像が4に分類された、という意味。 print(y_train.shape) # (60000,) データセットの選択 X_train,y_train、X_test,y_testから、値が5または8のものだけのViewを取得する。 そのために、まず値が5または8のものだけのインデックスを取得する。 NumPyのwhereはndarrayのうち条件を満たす要素のインデックスを返す。 X_trainに入っている60000件の2d arrayのうち、 値が5または8のインデックス(0-59999)を取得するのは以下。 index_train = np.where((X_train==5)|(X_train==8)) print(index_train) # (array([ 0, 11, 17, ..., 59995, 59997, 59999]),) index_test = np.where((X_test==5)|(X_test==8)) print(index_test) # (array([ 8, 15, 23, ..., 9988, 9991, 9998]),) インデックスを使って絞り込む。 X_train,y_train = X_train[index_train],y_train[index_train] X_test,y_test = X_test[index_test],y_test[index_test] print(X_train.shape) # (11272, 28, 28) print(X_test.shape) # (1866, 28, 28) 前処理 0-255の間の値を0-1の間の値に変換する（正規化）。 28x28の画像(2darray)を1x784(1darray)に整形する（平坦化)。 X_train,X_test = X_train / 255.0, X_test / 255.0 X_train = X_train.reshape(X_train.shape[0], X_train.shape[1] * X_train.shape[2]) X_test = X_test.reshape(X_test.shape[0], X_test.shape[1] * X_test.shape[2]) ベストなハイパーパラメータの選択 線形サポートベクトル分類器を作成する。 from sklearn.svm import LinearSVC linsvc = LinearSVC(loss=\"squared_hinge\",penalty=\"l1\",dual=False) 線形サポートベクトル分類器のハイパーパラメータCの選択 逆正則化パラメータCをGridSearchCVで探す。MBP2013Laterで学習(fit)に5分くらいかかった。 GridSearchCVからはC=0.2がbestと返ってくる。 from sklearn.model_selection import GridSearchCV param_grid = {\"C\":[0.025,0.05,0.1,0.2,0.4]} model = GridSearchCV(estimator=linsvc, param_grid=param_grid,cv=5,scoring=\"accuracy\",return_train_score=True) model.fit(X_train,y_train) print(model.cv_results_[\"mean_train_score\"]) # array([0.96291693, 0.96775192, 0.97059085, 0.97340754, 0.97626859]) print(model.cv.results_[\"mean_test_score\"]) # array([0.95626331, 0.95990064, 0.96158623, 0.9625621 , 0.96105394]) print(model.best_params_) # {\'C\': 0.2} 学習、精度評価 C=0.2を使って新しく学習させる。 linsvc = LinearSVC(loss=\"squared_hinge\",penalty=\"l1\",dual=False,C=0.2) linsvc.fit(X_train,y_train) 訓練データ、テストデータに対して正答率を求める。 訓練データについて97.2%、テストデータについて96.2%。 過学習すると訓練データが高くテストデータが低くなる。 from sklearn.metrics import accuracy_score pred_train = linsvc_best.predict(X_train) acc = accuracy_score(y_true = y_train,y_pred = pred_train) print(acc) # 0.9723207948899929 pred_test = linsvc_best.predict(X_test) acc = accuracy_score(y_true = y_test,y_pred = pred_test) print(acc) # 0.9619506966773848 モデルの解釈可能性 [mathjax] 線形SVMの決定境界(f(x))の係数をヒートマップっぽく表示して、どの係数を重要視しているかを確認する。 基本的に真ん中に画像が集まっているので、28x28の隅は使わないのが正しそう。 正則化パラメータによって係数の大きさを制御しているため、正則化パラメータを変えると係数が変わる。 今回のは(L_1)正則化なので、係数が0のものが増える..らしい（..別途調べる..)。 (f(x) = w_0 + w_1 x_1 + w_2 x_2 + cdots w_{784} x_{784}) import matplotlib.pyplot as plt weights = linsvc_best.coef_ plt.imshow(weights.reshape(28,28)) plt.colorbar() plt.show()

default eye-catch image.

決定木の分割基準,情報ゲイン,エントロピー

[mathjax] 集合に対して再帰的に境界を入れていく操作が決定木の作成。 では、集合のどこに境界を入れれば良いか。 属性をテストすることにより得られる情報量が最も大きくなるように入れる。 汎化能力、みたいな言葉を読んでいくにあたってこの先、結構抽象的な話になるので一度確認。 データが(n)個のクラスに分類できるとして、クラス(i)に属する確率を(P_i)とする。 このとき、あるデータがクラス(i)に属することを知るには(log_2 frac{1}{P_i})の情報量が必要。 その期待値は(I(P_1,P_2,cdots,P_n)=sum_{i=1}^{n} P_i log_2 frac{1}{P_i} = - sum_{n}^{i=1} P_i log_2 P_i)。 情報量の平均をエントロピーとか。 (D_p)個のデータを(D_{left})、(D_{right})に分割するとする。 その時、属性(f)に関する問いを使って分割する。 2分木の左ノード、右ノードだからleft,right。 (I(D_p))は分割前のエントロピー。 (I(D_{left}))、(I(D_{right}))は分割後のエントロピー。 分割前には(I(D_p))ビット必要だったのが、 分割後には(frac{N_{left}}{N_p} I(D_{left}) + frac{N_{right}}{N_p} I(D_{right}))ビット必要になった。と読むらしい。 その差を情報ゲイン(IG(D_p))と呼んで以下のように定義するらしい。 begin{eqnarray} IG(D_p,f) = I(D_P) - frac{N_{left}}{N_p} I(D_{left}) - frac{N_{right}}{N_p} I(D_{right}) end{eqnarray} 分割前よりも分割後の方が(IG(D_p,f))だけエントロピーが低い、という事実に関して、 分割に使った問いにより、(IG(D_p,f))の情報量を獲得した、と考えるらしい。 情報ゲインが最大になるような問いを根にもってきて、 再帰的に情報ゲインが大きいものから順に問うことで決定木を作っていく。