【データ可視化/Streamlit】StreamlitアプリからファイルをアップロードしてSnowflakeのテーブルを自由に更新する方法

Posted at 2024-03-17

前置き

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

最近、Streamlitでデータ可視化アプリを作成しており、
Streamlitアプリに行数がやや大きいデータをコピペして、
SnowFlakeのテーブルを自由に更新したい！という状況がありました。

しかし、Streamlitに容量のやや大きいデータを貼り付けると、
ブラウザが重くなったり固まってしまうという問題点が発生してしまいました。

今回はその解決策を考えてみました！

結論

データをCSVファイルに貼り付けて、そのCSVファイルと同じ内容にSnowFlakeのテーブルを更新する方法がとても良かったです。

ブラウザが固まることもなくなり、アプリの動きが一気に早くなりました。

機能を実装する

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

前準備

まずは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

こういったコードを作成しました。
CSVをデータフレーム化して、データフレームの内容でSnowflakeのテーブルを全件洗い替え更新しています。

test_app.py

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


# 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


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

uploaded_file = st.file_uploader("ファイルを選択 or ドラッグしてください。")
employees = get_dataset("EMPLOYEES")

if uploaded_file is not None:
    employees = pd.read_csv(uploaded_file, encoding="utf-8")

if st.button('Submit'):
    try:
        session.write_pandas(
            employees,
            "EMPLOYEES",
            overwrite=True,
            quote_identifiers=False
            )
        st.success("アップロードが成功しました")
        time.sleep(5)
    except Exception as e:
        st.warning(f"アップロードに失敗しました:{e}")
    st.experimental_rerun()
st.dataframe(employees, height=800)