BLOG 開発者ブログ

ホーム>
開発者ブログ>
Snowflake と Streamlit でデータアプリケーション開発入門

2022年12月12日

Snowflake と Streamlit でデータアプリケーション開発入門

今年6月に行われた Snowflake Summit でのさまざまな発表の中に、Streamlit の機能統合がありました。

将来的に Streamlit は Snowflake に統合され Snowflake ウェブ UI 上での開発が可能になるとのことで、これは重要な機能になるに違いないと思い、今回 Streamlit に入門してみようと思います！
（公開されているデモ動画）

はじめに

こんにちは。モバイルソリューショングループの arai.ya です。

この記事はアイソルート Advent Calendar 2022の12日目の記事です。昨日は kikuchi.s さんの「【dynamoose】DynamoDBから1MB以上のデータを取得する方法」でした。ぜひこちらもご覧ください。

Streamlit とは

Streamlit は、データに特化した Python 用ウェブアプリケーション開発フレームワークです。データサイエンスや機械学習向けのアプリケーションを、ウェブフロントエンド開発の経験がなくても簡単に構築することができます。

https://streamlit.io

また、今年の3月に Snowflake が Streamlit を買収しています。今後は Snowflake との機能統合が進み、 Snowflake が目指すデータアプリケーションの普及がどんどん加速していくのだと思います。
データアプリの民主化 – SnowflakeのStreamlit買収

それでは実際に、Streamlit と Snowflake を使って簡単なアプリケーションを作成してみます！

Snowflake サンプルデータの準備

世界中が今サッカーワールドカップで盛り上がっていますね！

ところで Snowflake には Data Marketplace という、自社のデータを外部に公開し提供することができる機能があります。その中に、「Opta Data: FIFA Men’s World Cup – SAMPLE」という無料で公開されているデータセットがあります。過去のワールドカップの試合詳細データ（の一部）が見られるようです。今回は、このデータを Steamlit を使って簡単に可視化してみます！

まずは以下から Snowflake トライアルアカウントを作成します（詳しい手順はここでは省略します）。
https://signup.snowflake.com/

Snowflake のウェブ UI が開いたら、Marketplace から、「Opta Data: FIFA Men’s World Cup – SAMPLE」を検索して取得します。

以下のように Opta_Data_FIFA_Mens_World_Cup__SAMPLE というデータベースが見えたらOKです。ぜひここで一旦、どのようなデータが含まれているかを色々と調べてみてください。

次は Streamlit を使う準備をします。以下は macOS 12.6.1 で実行しています。

開発環境の準備

今回の作業用ディレクトリを作成し、その中で作業します。

mkdir snowflake-streamlit-playground
cd snowflake-streamlit-playground

1 2	mkdir snowflake-streamlit-playground cd snowflake-streamlit-playground

Streamlit のインストール

今回はローカルに Streamlit をインストールしてアプリケーションを動作させます。Steamlit はこちらを参考にインストールします。

pipenv shell
pipenv install streamlit

1 2	pipenv shell pipenv install streamlit

Snowflake Python Connector のインストール

Streamlit から Snowflake に接続してデータを取得するために、Snowflake Python Connector を使用します。こちらを参考にインストールします。今回は Python 3.10.8 を使用しています。

pip install -r https://raw.githubusercontent.com/snowflakedb/snowflake-connector-python/v2.8.3/tested_requirements/requirements_310.reqs

1	pip install -r https://raw.githubusercontent.com/snowflakedb/snowflake-connector-python/v2.8.3/tested_requirements/requirements_310.reqs

Pandas DataFrame も使いたいので、Pandas API を提供しているバージョンをインストールします。

pip install "snowflake-connector-python[pandas]"

1	pip install "snowflake-connector-python[pandas]"

Snowflake 接続情報の準備

作業ディレクトリ直下に .streamlit/secrets.toml というファイルを作成し、以下の内容で保存します。

[snowflake]
user = "USER_NAME"
password = "YOUR_PASSWORD"
account = "SNOWFLAKE_ACCOUNT"
warehouse = "COMPUTE_WH"
database = "Opta_Data_FIFA_Mens_World_Cup__SAMPLE"
schema = "WORLD_CUP"

[snowflake]

user = "USER_NAME"

password = "YOUR_PASSWORD"

account = "SNOWFLAKE_ACCOUNT"

warehouse = "COMPUTE_WH"

database = "Opta_Data_FIFA_Mens_World_Cup__SAMPLE"

schema = "WORLD_CUP"

user, password, account, warehouse はそれぞれご自身の環境の値で読み替えてください。
（※ GitHub などでソースコードを管理する場合は、このファイルを .gitignore に追加してコミット・プッシュしないようにしましょう）

これで準備は完了です！さっそく Streamlit でアプリをつくりましょう！

Streamlit アプリ作成

app.py というファイルを作成して、まずは以下のように必要なモジュールのインポートと Snowflake からデータを取得する汎用的な関数を定義します。

import streamlit as st
import pandas as pd
import snowflake.connector


@st.experimental_singleton
def init_connection():
    return snowflake.connector.connect(
        **st.secrets["snowflake"], client_session_keep_alive=True
    )


conn = init_connection()


@st.experimental_memo(ttl=600)
def run_query(query):
    with conn.cursor() as cur:
        cur.execute(query)
        return cur.fetch_pandas_all()

import streamlit as st

import pandas as pd

import snowflake.connector

@st.experimental_singleton

def init_connection():

return snowflake.connector.connect(

**st.secrets["snowflake"], client_session_keep_alive=True

)

conn = init_connection()

@st.experimental_memo(ttl=600)

def run_query(query):

with conn.cursor() as cur:

cur.execute(query)

return cur.fetch_pandas_all()

Streamlit では、タイトルや文章を簡単にそれっぽく表示させることができます。
以下の記述を追加します。

st.title("2018 FIFA World Cup Sample Data Visualization")

1	st.title("2018 FIFA World Cup Sample Data Visualization")

ここでアプリケーションを実行してみます。

streamlit run app.py

1	streamlit run app.py

タイトルがそれっぽく表示されました！

次に、試合のデータを取得して表示してみます。以下を追加して、Streamlit アプリ画面右上の「Rerun」ボタンを押してみます（ちなみに Always rerun を押すと、ファイルの保存時に勝手に再実行してくれるので便利です）。

df_games = run_query(
    """
    select
        g.id as game_id,
        g.season_id as season_id,
        v.city as city,
        v.name as venue,
        ht.name as home_team_name,
        at.name as away_team_name,
        substr(g.game_date::string, 1, 10) as game_date,
        g.home_score as home_score,
        g.away_score as away_score
    from
        "Opta_Data_FIFA_Mens_World_Cup__SAMPLE"."WORLD_CUP"."GAME" as g
    left join
        "Opta_Data_FIFA_Mens_World_Cup__SAMPLE"."WORLD_CUP"."VENUE" as v
        on g.venue_id = v.id
    left join
        "Opta_Data_FIFA_Mens_World_Cup__SAMPLE"."WORLD_CUP"."TEAM" as ht
        on g.home_team = ht.id
    left join
        "Opta_Data_FIFA_Mens_World_Cup__SAMPLE"."WORLD_CUP"."TEAM" as at
        on g.away_team = at.id;
"""
)

df_games.set_index("GAME_ID", inplace=True)

st.write(df_games)

df_games = run_query(

"""

select

g.id as game_id,

g.season_id as season_id,

v.city as city,

v.name as venue,

ht.name as home_team_name,

at.name as away_team_name,

substr(g.game_date::string, 1, 10) as game_date,

g.home_score as home_score,

g.away_score as away_score

from

"Opta_Data_FIFA_Mens_World_Cup__SAMPLE"."WORLD_CUP"."GAME" as g

left join

"Opta_Data_FIFA_Mens_World_Cup__SAMPLE"."WORLD_CUP"."VENUE" as v

on g.venue_id = v.id

left join

"Opta_Data_FIFA_Mens_World_Cup__SAMPLE"."WORLD_CUP"."TEAM" as ht

on g.home_team = ht.id

left join

"Opta_Data_FIFA_Mens_World_Cup__SAMPLE"."WORLD_CUP"."TEAM" as at

on g.away_team = at.id;

"""

)

df_games.set_index("GAME_ID", inplace=True)

st.write(df_games)

Snowflake から取得したデータを簡単に表形式で表示できました！

st.write(df_games) の代わりに以下のように書くことで、試合一覧からプルダウンで選択して、選択した試合内容を表示します。

selected_game_index = st.selectbox(
    "表示する試合を選択：",
    options=df_games.index,
    format_func=lambda x: f'{df_games.loc[x]["HOME_TEAM_NAME"]} vs. {df_games.loc[x]["AWAY_TEAM_NAME"]}',
)

st.header("試合詳細")
st.text(f'日程： {df_games.loc[selected_game_index]["GAME_DATE"]}')
st.text(
    f'会場： {df_games.loc[selected_game_index]["VENUE"]}, {df_games.loc[selected_game_index]["CITY"]}'
)
st.text(
    f'スコア： {df_games.loc[selected_game_index]["HOME_SCORE"]} - {df_games.loc[selected_game_index]["AWAY_SCORE"]}'
)

selected_game_index = st.selectbox(

"表示する試合を選択：",

options=df_games.index,

format_func=lambda x: f'{df_games.loc[x]["HOME_TEAM_NAME"]} vs. {df_games.loc[x]["AWAY_TEAM_NAME"]}',

)

st.header("試合詳細")

st.text(f'日程： {df_games.loc[selected_game_index]["GAME_DATE"]}')

st.text(

f'会場： {df_games.loc[selected_game_index]["VENUE"]}, {df_games.loc[selected_game_index]["CITY"]}'

)

st.text(

f'スコア： {df_games.loc[selected_game_index]["HOME_SCORE"]} - {df_games.loc[selected_game_index]["AWAY_SCORE"]}'

)

このように、ユーザーの操作を受け付けて動的に表示内容を変えることもできます。今回は簡単な例だけしかご紹介できませんが、他にもさまざまな図表を表示することができるようなので、また次回にでも挑戦してみようと思います！

おわりに

実際に Streamlit を触ってみると、確かに簡単にデータを扱うアプリケーションが作れると感じました。どこまで複雑なことができるかはもう少し調べてみる必要がありますが、サクッとデータを共有するためのプロトタイピングツールとしても使いやすそうです。今後の Snowflake との機能統合がとても楽しみです！

おまけ

サッカーの試合詳細データをプロットするライブラリを見つけたので、組み合わせて使ってみます。
mpl-soccer

まずはライブラリをインストールします（ここまでと同じ pipenv 仮想環境と作業ディレクトリで）。

pip install mplsoccer

1	pip install mplsoccer

app.py で mpl-soccer をインポートします

from mplsoccer.pitch import Pitch

1	from mplsoccer.pitch import Pitch

以下を追記します。

st.header("Ball Event Data")

df_event_types = run_query(
    'select id, name from "Opta_Data_FIFA_Mens_World_Cup__SAMPLE"."WORLD_CUP"."EVENT_TYPE" order by id;'
)
df_event_types.set_index("ID", inplace=True)

selected_event_type_index = st.selectbox(
    "Event type を選択：",
    df_event_types.index,
    format_func=lambda x: df_event_types.loc[x]["NAME"],
)

df_events = run_query(
    f"""
    select
        x, y
    from
        "Opta_Data_FIFA_Mens_World_Cup__SAMPLE"."WORLD_CUP"."EVENT"
    where
        event_type_id = {selected_event_type_index}
        and
        game_id = {selected_game_index};
    """
)


pitch = Pitch(pitch_type="opta")
fig, ax = pitch.draw()
sc = pitch.scatter(df_events.X, df_events.Y, ax=ax)
st.pyplot(fig)

st.header("Ball Event Data")

df_event_types = run_query(

'select id, name from "Opta_Data_FIFA_Mens_World_Cup__SAMPLE"."WORLD_CUP"."EVENT_TYPE" order by id;'

)

df_event_types.set_index("ID", inplace=True)

selected_event_type_index = st.selectbox(

"Event type を選択：",

df_event_types.index,

format_func=lambda x: df_event_types.loc[x]["NAME"],

)

df_events = run_query(

f"""

select

x, y

from

"Opta_Data_FIFA_Mens_World_Cup__SAMPLE"."WORLD_CUP"."EVENT"

where

event_type_id = {selected_event_type_index}

and

game_id = {selected_game_index};

"""

)

pitch = Pitch(pitch_type="opta")

fig, ax = pitch.draw()

sc = pitch.scatter(df_events.X, df_events.Y, ax=ax)

st.pyplot(fig)

試合中のパスなどのイベントの位置をプロットすることができました。
ここまでの応用で、イベントの種類を選択して表示を切り替えています。

サッカーの試合中のさまざまなデータの可視化が簡単にできそうです！

ブログ記事一覧へ戻る

arai.yaのブログ

ブログ内検索