はじめに

本記事は、Snowflake Advent Calendar 2024 の2日目の記事となります。昨日の記事は、ak-sakatoku さんの Snowflake Data Superhero としての1年間を振り返る記事でした。イベント立ち上げや運営など、Sakatoku さんがモリモリと進めていく様子を横目に見ていられたのは幸運でした！私も見習って、積極的に活動できるようになりました。

というわけで今回の記事は、私が初めて主体的に携わったイベントについてのお話です。かれこれ3ヶ月も前の話になってしまいましたが、2024年9月12日の Snowflake World Tour でコミュニティとして枠をいただき、#SnowVillage の一員でイベントを開催してきた話（開発編）についてご紹介していこうと思います。

イベントでは Snowflake に関するクイズを Streamlit アプリで提供したので、その内容について徹底解剖を試みます。今回開発したアプリは、なるべく Streamlit の一般的な機能を使うようにしつつ、多くの機能を詰め込むことを目指したため、Streamlitの限界に挑んだと言っても過言ではないかもしれません。

なお、アプリやソースコードへは下記からアクセスいただくことができます。
ぜひアプリを触りながら記事を読んでいただけますと幸いです！

通常問題

最終問題

背景

昨年のイベントではスライドでクイズ運用をしていたが、今年は会場がホール型の大きい箱一つだったこともあり（＋Streamlit 好きが多かったこともあり）アプリでの非同期的なクイズ運用に挑戦した
開発期間はちょうど1ヶ月
イベント参加者は200名ほど
アプリは、下記のような形でスマホからアクセスしてもらう形を取った
私はこのアプリの実装班（アプリ開発、クイズ作問）のリーダーとして全体統括・ストーリー班側の思いを具現化する役割を担っていました。
アプリ開発班は、私含め5名でした。私がアプリの基盤やバックエンド、APIを開発し、その後、クイズ自体やデザインなどを協力して実装していきました。

本記事の構成

本記事ではまず、アプリの概要としてウォークスルーを行っていきます。その後、実装に困った部分や悩んだ部分、工夫した部分を紹介します。

さらに、今後の課題について考察します。1ヶ月しか開発期間がなかったこともあり、基底部分には大きな変更が入れられませんでした。そこで、改めて改善できそうなポイント、今ならこう実装できそうだといったポイントを整理しておきます。

アプリの概要

アプリ全体は、次のように構成されます。

通常問題でのログイン画面
通常問題のクイズ画面
通常問題の回答状況画面
最終問題（3個）

通常問題のクイズは8つあり、それらをある基準までクリアすると最終問題に進むことができるようになっています。最終問題には、各チームから1名ずつ代表者を選出いただき、前方のスクリーンで一体となって挑戦いただくクイズを用意しました。

基盤

今回のアプリ基盤は、Streamlit Community Cloud を、データベースや各種 LLM などの機能については Snowflake を利用しました。

下図にアーキテクチャを示します。

Streamlit Community Cloud では、各アプリのリソース制限が1GBと公表されています。

そのため、参加者全員が一つのアプリ URL にアクセスするのではなく、いくつかのチームが1つのアプリを使用するように、複数の同一内容のアプリを動作させていました。

通常問題　ログイン部分（app.py）

トップページには、ログインページを用意しました。と言ってもこの画面自体で認証を行っている訳ではなく、会場入場時にランダム配布したカードへ記載されたチームIDを選択することで、自動的に Snowflake との接続が行われるようになっています。

チームが選択されると、「挑戦を開始する」ボタンが表示され、このボタンを押すことでクイズ画面に遷移できるようになっています。

ソースコードを見ていただくと分かりますが、ログイン処理以外は、ここではスタイルの適用やテキストの表示を行っているだけです。
https://github.com/THiyama/SWTT-Community/blob/main/app.py

ログイン処理については、下記のようにシンプルな実装となっています。

team_id = st.selectbox(
    label="結成するチームを選択",
    options=list(TEAMS.keys()),
)

if team_id:
    st.session_state.clear()
    st.session_state.team_id = team_id
    st.session_state.snow_session = create_session(TEAMS[team_id])

    if st.button("挑戦を開始する"):
        print(st.session_state)
        st.switch_page("pages/01_normal_problems.py")

改めてソースコードを読んで気が付きましたが、st.selectbox のデフォルト値を表示しないために、TEAMS 定数（辞書型）の一要素目に "" という空白の要素を入れていました。が、これは options=None を設定するだけで済みますね。

また、辞書型のため順序が保証されないはずなので、この実装だとデフォルト値が意図しない値になる可能性があるのでその点でも微妙っぽいです。

通常問題　クイズ部分

続いて、画面遷移後のクイズ画面の紹介です。

参加者の皆様が最も長く触る部分ということもあり、メンバー一同の魂がこもった場所です。問題自体の実装もそうですし、文面、デザイン、レスポンス、仕組みなどなど、とにかく全員でこだわり抜きました。

このページでは、クイズの選択とそのクイズへの回答を行います。

クイズの回答

クイズを提出すると、次のような成功・失敗メッセージとともに「クリスタルと通信中」というテキストが表示されます。

該当のコードは次のようなものになります。ある程度時間のかかる処理はこのように st.spinner などを活用しています。

with st.spinner("クリスタルと通信中..."):
    snow_df = session.create_dataframe(df)
    snow_df.write.mode("append").save_as_table("submit2")

ここでは、Snowflake テーブルへの回答状況の保存を行っています。この情報は、各クイズに対する正解状況の可視化や、回答制限のカウントに利用しています。（回答制限に関しては少し厳しくしすぎて本番中に何度か失敗状況をリセットしたので、適切な閾値を検討しておくべきだったと反省）

クイズの選択

クイズ選択の UI は次のようにセレクトボックスで実装しました。各チームの回答状況を、別セッション（同じチームの別のスマホ）からでも把握できるようになっています。

通常問題　回答状況の表示部分

アプリの中でこの画面を見ることもできますが、基本的にはスクリーンに投影し KT に実況してもらうことをイメージしていました。

実際、上図のようにこのクイズ全然解けてないから解いてー！と呼びかけたり、このクイズ難しすぎたか・・・！と作問者が頭を抱えたりなど、コンテンツに彩りを添えるような機能にできたかなと思っています。リアルタイムにピョコピョコとグラフが更新されるのも楽しかったですね。

また、グラフに赤いラインが引いてありますが、これが先述の最終問題に進むことができる最低ラインです。赤いラインを超えたクイズの描画に関しては、自動的に薄い水色で表現するようにしています。

まさにこれはデータアプリだな。とスクリーンに映し出されたこの画面を見て思いました。

最終問題

最終問題は、めちゃくちゃシンプルな実装になっています。通常問題の API や仕組みを活用して実装しています。

最終問題のトップページは、下記のようにシンプルな実装となっています。

st.title("💎データクリスタルの復活")

css_name = apply_default_custom_css()
message = f"""
    この旅もいよいよ終盤です。<br>

    <strong>選ばれしものたちよ、いざ最後の挑戦に立ち向かうのです！</strong>
    """

display_applied_message(message, css_name)

if st.button("挑戦を開始する"):
    st.switch_page("pages/final1.py")

また、問題に関しても画像の表示やラジオボタンによる選択、テキストの入力などに留まっています。

実装時は、ストーリー班の要望に応じてインタラクティブにアプリを構築していきました。

ここはこういう風なヒントを出そう→こんな感じね！といった感じで、ビジネス側の要求にすぐさま応えられるのも Streamlit の良いところだなーと感じながら実装していたことを今も鮮明に記憶しています。

【全体】実装のこだわりポイント

アプリ・基盤の全体像をご紹介したところで、こだわりポイント・困難だったポイントなどを紹介していこうと思います。

クイズ API とテンプレートの提供

先述した通り、クイズ自体の実装は私含む5名で行いました。そのため、クイズ描画・処理に統一感を出すために、クイズ API をいくつか用意し、Streamlit や Python が書ける人であれば誰でも実装ができるようなテンプレ実装を提示しました。

実際に、クイズを実装するために必要なスクリプトは下記のみです。

def present_quiz(tab_name: str, max_attempts: int) -> str:
    st.header("Be Positive!", divider="rainbow")

    display_problem_statement("<問題文(省略)>")
    st.write(f"回答回数の上限は {max_attempts}回です。")
    answer = st.text_input("ポジティブメッセージを入力してください:", key=f"{tab_name}_answer")

    return answer

def process_answer(answer: str, state, session: Session) -> None:
    correct_value = 0.8
    answer_sentiment = Sentiment(
        Translate(answer, "ja", "en", session=session), session=session
    )
    if answer_sentiment > correct_value:
        state["is_clear"] = True
        st.success(f"ポジティブ！あなたのポジティブは{answer_sentiment}でした。")
    else:
        state["is_clear"] = False
        st.error(
            f"ポジティブが足りません！あなたのポジティブは{answer_sentiment}でした。0.8 を超えるように頑張りましょう！"
        )

    save_table(state, session)

ね、めちゃめちゃ簡単でしょう？これだけで、この画面の機能を実現できます。Streamlit がすごいという話でもありそうです。

この画面の中だけでも、カッコいい感じに問題文を表示する、回答の正否を判断して Snowflake に書き込む、状態を表示する、成功・失敗を判定する、といった仕組みが組み込まれています。

上記の仕組みについては、こちらの run 関数に実装されていますが、実装者は気にする必要はありません。

def run(tab_name: str, session: Session):
    state = init_state(tab_name, session, MAX_ATTEMPTS_MAIN)
    main_attempt = init_attempt(
        max_attempts=MAX_ATTEMPTS_MAIN, tab_name=tab_name, session=session, key="main"
    )

    answer = present_quiz(tab_name, MAX_ATTEMPTS_MAIN)  # ★

    placeholder = st.empty()
    if check_is_failed(session, state):
        process_exceeded_limit(placeholder, state)
    elif placeholder.button("submit", key=f"{tab_name}_submit"):
        if main_attempt.check_attempt():
            if answer:
                process_answer(answer, state, session)  # ★
            else:
                st.warning("選択してください")

        else:
            process_exceeded_limit(placeholder, state)

    clear_submit_button(placeholder, state)

ただ、今にして思うとクラスの継承を利用していれば諸々を、もう少し簡単に実装できたような気がしますね。

クイズ選択 UI の実装「st.tabs vs st.selectbox」

最終的には、st.selectbox を採用することになったのですが、当初は st.tabs を使用していました。

こんな感じです。結構カッコいいんですよね。

しかし、次のような問題がありました。

分かりますか。このもっさり感。

ちなみに、理想（本番）はこんな感じです。

クイズ実装自体がアップデートされているため若干挙動は違いますが、レスポンス自体は一目瞭然ですよね。何が違うのでしょうか。

それが、先述したように st.tabs の特徴によるものでした。それは、st.tabs がタブに並べている機能はすべて再描画をしているという特徴です。これはこれでメリットがあり、タブ間の移動がスピーディになります。しかし、今回のアプリではどうでしょうか？クイズ間の移動が高速である必要はありませんよね。

要は選択の問題なのですが、私は当初この違いを意識できていなかったため、誤った選択をしてしまっていました。

なお、実装については意外とシンプルに差し替えに成功しています。

python

- selected_tab = st.tabs(tab_titles)

- for i, tab_title in enumerate(problem_ids):
-     with selected_tab[i]:
-         tabs[tab_title].run(tab_title, session)

+ selected_problem = st.selectbox("挑戦する問題を選択してください", options=tab_titles)
+ selected_problem_id = problem_ids[tab_titles.index(selected_problem)]
+ tabs[selected_problem_id].run(selected_problem_id, session)

（タブで実装していたときの変数名が残ってしまっていますね）

これらの特徴を頭の片隅に入れておくと、今後選択の可能性が生じた際や、パフォーマンスに問題が生じた際に適切な方法を選択できると思います。

パフォーマンスの改善

先述のクイズ選択 UI に関してもですが、アプリ実装中はどの処理が重いのかさっぱり分かりませんでした。そこで一念発起し、Streamlit でのパフォーマンスチューニングについて調査したところ、streamlit-profiler といった Streamlit カスタムコンポーネントがあることを知りました。使い方は非常にカンタンです。

from streamlit_profiler import Profiler

with Profiler():
    # Streamlit の処理

これだけで、次のようなプロファイラが Streamlit の画面下部に表示されるようになります。

このプロファイラを眺めて、無駄な処理や重い処理、そもそも処理ではなく描画待ちが発生している、などに気がつくことができました。

クイズ選択 UI における回答状況の表示

正解した問題には✅️を、失敗した問題には❌️をつけていた箇所についてです。

基本的にはその状況を取得する API やクイズ名に絵文字を付加するだけなのですが、セレクトボックスのオプション文字列を変更すると、セレクトボックスで選択されているオプションが一番上のものに戻ってしまいます。

検証コード

import streamlit as st

options = ["a", "b", "c"]

if st.toggle("絵文字を追加"):
    options[1] += "❤"

st.selectbox("選択", options=options)

要は、問題に正解／失敗するたびにクイズ選択が一番はじめの問題に戻されてしまうのです。これは微妙ですよね。

そこで私に残された選択肢は2つです。クイズ選択に絵文字を付加することを諦めるか、なんとかこの問題を解消するか。こういうときは楽しそうな方を選びましょう。

という訳で、実現できました！

ソースコードとしては下記のようになります。8行目以降を追加・編集する必要があります。

import streamlit as st

options = ["a", "b", "c"]

if st.toggle("絵文字を追加"):
    options[1] += "❤"

if "selected_index" not in st.session_state:
    st.session_state.selected_index = 0

def update_selected_index():
    st.session_state.selected_index = options.index(
        st.session_state.selected_option
    )

st.selectbox(
    "選択", 
    options=options,
    index=st.session_state.selected_index,
    on_change=update_selected_index,
    key="selected_option"
)

やっていることは単純で、コールバック関数を使って、現在選択されているオプションを selected_index として取得しておきます。そして、st.selectbox の index オプションにその値を格納することで、選択されているオプションを前回の値のままにすることができます。

他にもいい実現方法があれば教えて下さい。また、もしかすると、2025年4月までにはこういうことをしなくても良くなるかもしれません。

🗽Accept new options in st.selectbox and st.multiselect 👟Planning

回答状況におけるリアルタイム描画

これは、みんなだいすき st.fragment により実現しています。

@st.fragment(run_every="10s")
def update_chart():
    # 処理

update_chart()

上記のようにデコレータをつけるだけで、10秒ごとに当該関数のみ再実行が走るようになります。呼び出すときは、普通の関数として呼び出すだけでOKです。

回答状況におけるクリアしたクイズの強調と雪を降らせる処理

特に難しいことはしていませんが、ちょっとした工夫ということで紹介しておきます。

回答状況を Snowflake から取ってきて描画している訳ですので、容易に回答状況が取得できます。あとは適当な閾値を設けて、その値を各クイズが上回ったらグラフの色を変えたり、雪を降らせたりするだけです。

コードとしては次のようなものになります。

result["color"] = "#29B5E8"  # デフォルトカラー（薄いSnowflake色）

# 雪を降らせる処理
for index, row in result.iterrows():
    if row["IS_CLEAR"] >= CLEAR_COUNT:
        result.at[index, "color"] = "#c2e5f2"  # 色を薄い青色に変更

        # クリアカウントを超えた場合の通知の表示と雪を降らせる処理
        if not st.session_state[f"{row['problem_id']}_is_over_clear"]:
            st.success(f"「{row['problem_name'][:-1]}」の的屋が解放されたようだ！")
            st.snow()
            st.session_state[f"{row['problem_id']}_is_over_clear"] = True

ここで注意したいのは、雪は閾値を超えた一回だけで良いので、クリアフラグをセッションステートに持たせていることです。

これだけの工夫で、演出としてはいい感じになるので、実装してよかった機能、そして愛着のある機能の一つです。

（この回答状況のリアルタイム表示ページ自体が愛着あったりはするのですが）

通常問題のフロー制御

スマホからだとあまり意識することはなかったかもしれませんが、誤ってチーム選択をせず問題画面に行ってしまっても、ちゃんと戻されるようにしていました。

もちろん、ステータスによって、表示されるエラーメッセージや画面遷移ボタンも変わります。

実装は非常に単純で、セッションステートに該当の変数が存在しない場合はエラーメッセージと画面遷移用のボタンを表示した後、st.stop() で後続の描画を停止するだけです。

def get_team_id():
    if "team_id" not in st.session_state:
        st.warning("そなたらは、まだチームとして誓いが結ばれていないようだの・・・。")
        if st.button("チーム結集に戻る"):
            st.switch_page("app.py")
        st.stop()
    else:
        return st.session_state.team_id

また、改めて見て気が付きましたが、下記のコードは例外により同様の処理を実装しており、一貫性がなくイケてないですね。上記のコードと同様に if 文で変数の有無を確認すれば良さそうです。

try:
    problem_ids = st.session_state.problem_ids
except AttributeError as e:
    st.warning("一度挑戦の場を訪れるが良い。")
    if st.button("挑戦の場に行く"):
        st.switch_page("pages/01_normal_problems.py")
    st.stop()

Snowflake セッションの持続時間の制御

意外と苦労したのがこの部分です。何かというと、アプリから Snowflake に接続したあと、一定時間（数時間）経過後にそのユーザーで接続できなくなってしまう問題が生じてしまいました。

厄介なのが、長時間経たないと問題が再現しないことで、改善したと思ったらまだその不具合が残っていた、という状況に陥りました。

そこで、しっかりと仕様を調査したところ、Snowflake ではセッションのタイムアウトが設けられており、デフォルトでは4時間と設定されていると分かりました。
https://docs.snowflake.com/ja/user-guide/session-policies#session-policies

そこで、セッション作成時に TTL（1時間）を設定し、問題が再発しなくなったことを確認できました。（しかし気付いたのが本番直前だったこともあり、本番には適用しませんでした。仮に発生したら該当チームのアプリを再起動する運用を運営メンバーの一部に伝えることで対応していました。）

@st.cache_resource(ttl=3600)
def create_session(team_id: str, is_info: bool = True) -> Session:
    session = st.connection(team_id, type="snowflake", max_entries=1).session()
    return session

今後の改善点

セッションステートの乱立

ここまで規模の大きいものを無邪気に作ると、セッションステートの管理が辛くなりますね。開発の佳境にセッションステートの設計を一新してやり直す時間もなく、結果的に針に糸を通すような繊細なセッションステート管理が必要になってしまいました。

今回はセッション変数を直書きしてしまったのですが、せめて列挙体やデータクラスを使ってセッションステートを用途ごとにグループ化する・エディタの補完が効くようにすることで、もっと楽になっていたのではないかと思います。

イメージとしてはこんな感じですね。そこまで労力をかけず、管理が楽になるのではないかと思います。

from enum import enum

class SessionKey(Enum):
    SNOW_SESSION = auto()
    PROBLEM_ID = auto()
    TEAM_ID = auto()

snow_session = st.session_state[SessionKey.SNOW_SESSION.name]
...

その他、今回のアプリではチームID・問題IDごとの回答状況を持っていました。これにより、クイズ選択UIにおける回答状況の可視化を実現していました。この場合は、静的な情報を保管する Enum ではなく、動的な情報を保管するデータクラスが適しているでしょうか。

from dataclasses import dataclass

@dataclass
class ProblemState:
    problem_id: str
    team_id: str

    @property
    def is_clear_key(self):
        return f"{self.problem_id}_{self.team_id}_is_clear"

    @property
    def is_failed_key(self):
        return f"{self.problem_id}_{self.team_id}_is_failed"

problem_state_key = ProblemState(problem_id = st.session_state[SessionKey.PROBLEM_ID],
                                 team_id = st.session[SessionKey.TEAM_ID])
                                 
is_clear = st.session_state[problem_state_key.is_clear_key]
is_failed = st.session_state[problem_state_key.is_failed_key]

これなら、毎回セッションステートのキー名を覚えることなくエディタの補完機能に委ねることができそうです。他に良い方法があったらご教示ください。（GitHub で調べても Enum を使っているケースがなく不安。便利そうなんだけどな。）

書き込み性能の改善

Snowflake は分析テーブルのため、アプリ用途として考えると書き込み性能が心許ないです。

そこで、Snowflake は Hybrid Tables というトランザクションにも対応できる HTAP データベースを用意しています。これ自体は相当簡単に使うことができるのですが、残念ながら今回使用した Snowflake トライアル環境では使用することができず、採用を見送りました。（シクシク）

一方で、やはりこうしたユースケースにおいて、Hybrid Tables の有用性が体感できたことは非常に有益な経験でした。

アプリ基盤をオートスケールするものにする

タイトル通りですね。時間と予算があれば、AWS ECS などでも実装してみたいです。もし Snowpark Container Services がスマホからのアクセスにネイティブ対応したら、そっちの方も楽しそうです。

トランザクション制御を導入（検証）する

実は、本番中にやらかしました。本文中にも記載しましたが、失敗の制限回数が厳しかったこともあり、どうやら回答制限で回答できなくなっているチームが多そうだということが分かります。

それも想定の範疇で、失敗カウントをリセットするスクリプトを用意していました。

そして意気揚々と失敗カウントをリセット・・・！
したところ、何故か成功カウントまで大幅に減ったクイズがある・・・😇

そのときは別に用意していた成功カウントを少しお化粧するクエリを実行し即座に元の状態に戻すことが出来たのですが、流石にヒヤヒヤものでした。（ちょうど全クイズのしきい値が目標値を上回りそうなクライマックスだったこともあり・・・）

しっかり検証した訳ではないのですが、運用クエリ自体にはバグらしき部分はなく、想定どおり動作します。そこで怪しそうなのが、書き込みと削除が大量に走ったことによる、トランザクションの不整合です。

単純に、運用スクリプトにトランザクションを張るだけで解決する問題かもしれませんが、ここはぜひとも検証しておきたいポイントです。

おわりに

この記事で、すべてとはいかないまでも、おおよそどのような処理が裏側で走っていたかその深淵を覗くことができたのではないでしょうか。もしほかに気になる箇所・挙動がありましたら、ぜひコメントください。暇なときに追記するようにします！

また、今更ですが、コミュニティイベントにご参加いただいた皆様、本当にありがとうございました。コミュニティとして Snowflake Love な皆様で横のつながりができていると嬉しいです。

今回の縁は、きっと次回以降のコミュニティイベント参加時に活きてくると思います！ぜひ、引き続きコミュニティイベントへ遊びに来て、徐々にコミュニティの良さを実感していっていただければと思います。直近では、SnowVillage, Japan - Build Meetup というこれまたドデカイイベントが開催されます。

最後に、コミュニティイベントをしっかりと楽しんでいただけたことを伝えてくださった皆様もありがとうございます。メンバー一同、思いを込めて準備を頑張ったので、糧になりました！