【データ可視化/Streamlit】Streamlitで作ったアプリからSnowflakeのテーブルを自由に更新する方法

前置き

こんにちは。データエンジニアの山口です！

Streamlitでデータ可視化アプリを作成しており、
Streamlitアプリ上に手入力した値とSnowflake内の値を結合してデータを可視化したいなと思ったので、やり方を考えてみました。

結論

Streamlitアプリ上に手入力した値をSnowflakeのテーブルにデータを挿入・更新して、
すでにSnowflakeに入っているデータと結合すればいいのではないかと言う考えに至りました。

Streamlitのform_submit_button関数が使えそうだったので、そちらを使っていきます！

機能を実装する

早速機能を実装していきます！

前準備

まずはStreamlitから更新をするテーブルを用意しておきます。

今回は従業員マスターというテーブルを以下のクエリで作成して、
このテーブルの中にINSERT文などで、いくつか適当にデータを入れておきます。

create or replace TABLE DEV_YAMAGUCHI_DBT.DATA_CATALOG_LT.EMPLOYEES (
	EMPLOYEE_ID NUMBER(38,0) COMMENT '従業員の氏名。このテーブルの主キー。',
	FIRST_NAME VARCHAR(50) COMMENT '名前。',
	LAST_NAME VARCHAR(50) COMMENT '苗字。',
	SALARY NUMBER(10,2) COMMENT '給与。ドル単位で格納。',
	HIRE_DATE DATE COMMENT '雇用した日。日本時間にて格納。'
)COMMENT='従業員マスター'
;

Streamlitでアプリを作る

テーブルを用意できたので、Streamlitでアプリを作っていきます。

1)ディレクトリを用意する

こういった構成でディレクトリを作ります。

root/
├── creds.json    # Snowflakeへの接続情報
└── test_app.py   # Streamlitのコード

2)creds.json

Snowflakeへの接続情報をjson形式で用意しておきます。

{
    "account": "[ACCOUNT名]",
    "user": "[USER名]",
    "password": "[PASSWORD]",
    "warehouse": "[WAREHOUSE名]",
    "database": "[DATABASE名]",
    "schema": "[SCHEMA名]",
    "role": "[ROLE名]"
}

3)test_app.py

こういったコードを作成しました。
Snowflakeからテーブルのデータを抽出して、
それらを編集してテーブルの更新をできるようにしてあります。

import streamlit as st
from snowflake.snowpark import Session
import json
from datetime import time

# creds.jsonからSnowflakeの接続情報を取得
def get_config_creds():
    with open('creds.json') as f:
        connection_parameters = json.load(f)
    return connection_parameters


# データセットを取得する
def get_dataset(table):
    df = session.table(table)
    return df


# テーブルを更新する
def table_submit(table_data, table_name, caption):
    with st.form(table_name):
        st.caption(caption)
        edited = st.data_editor(table_data, use_container_width=False, num_rows="dynamic")
        submit_button = st.form_submit_button("Submit")

    if submit_button:
        try:
            # Snowflakeに書き込み
            session.write_pandas(edited, table_name, overwrite=True, quote_identifiers=False)
            st.success("Table updated")
            time.sleep(5)
        except Exception as e:
            st.warning(f"Error updating table:{e}")
        st.experimental_rerun()
    return edited


# セッションを作成
connection_parameters = get_config_creds()
st.session_state.snowflake_connection = Session.builder.configs(connection_parameters).create()
session = st.session_state.snowflake_connection

# 作成したテーブルを変数に入れる
data = get_dataset("EMPLOYEES")

# データを更新するためのフォームを作成
with st.form("EMPLOYEES"):
    employees = st.data_editor(
        data,
        use_container_width=False,
        num_rows="dynamic",
        height=600
        )
    submit_button = st.form_submit_button("Submit")

# submitボタンを押したらEMPLOYEESテーブルに更新が走る
if submit_button:
    try:
        session.write_pandas(
            employees,
            "EMPLOYEES",
            overwrite=True,
            quote_identifiers=False
            )
        st.success("Table updated")
        time.sleep(5)
    except Exception as e:
        st.warning(f"Error updating table:{e}")
    st.experimental_rerun()

以上でアプリの作成が完了です！

アプリを使用してみる

$ streamlit run test_app.pyを実行して、実際にアプリを使っていきます。

コマンドを実行するとブラウザ上に以下のように表示されて、
自由にデータフレームを編集できるようになりました！

手入力でデータフレームを書き換えてSubmitボタンを押すと、

Snowflakeのテーブルが、書き換えたデータフレームと同じ内容に更新されています！

まとめ

以上でStreamlitからSnowflakeのテーブルを自由に更新することができるようになりました。

データエンジニア以外の方でも自由にStreamlit上からSnowflakeのテーブルを更新できるので、
手入力した値とSnowflakeのデータと掛け合わせてデータ分析をしたい時も便利です！

採用

GENDAデータチームではプロダクトのデータ解析や機械学習プロジェクトを推進できるデータサイエンティストを募集しています。
興味を持っていただけましたら、是非ご連絡いただけると幸いです！

参考記事