❄️
Streamlit in Snowflake でファイルのアップロードとダウンロードをしよう
はじめに
ついに Streamlit in Snowflake において st.file_uploader が利用可能となりました！
st.file_uploader は手元の PC から Streamlit 上にファイルをアップロードできるウィジェットを配置する機能であり、例えば CSV ファイルをアップロードしてデータフレームとして処理したり、Snowflake 上のテーブルにデータを保存したりすることが可能です。
今回はこの st.file_uploader を使って Snowflake 上のステージにファイルを格納したり、ステージ上のファイルから署名付き URL を発行するアプリをご紹介したいと思います。
機能概要
機能一覧
- ファイルをアップロードするステージを作成する機能
- ファイルをステージにアップロードする機能
- 1ファイル最大 200MB まで
- ファイル名にマルチバイト文字が含まれている場合も対応
- ステージにアップロードしたファイルの署名付き URL を 1 - 7 日の期間を設定して発行する機能
- ステージにアップロードしたファイルをダウンロードする機能
完成イメージ
ファイルアップロードとプレビュー画面
署名付き URL 発行画面
ファイルダウンロード画面
前提条件
- Snowflake アカウント
- python 3.11 以降 (特に追加のパッケージは不要です)
手順
新規で Streamlit in Snowflake のアプリを作成
Snowsight の左ペインから『Streamlit』をクリックし、『+ Streamlit』ボタンをクリックし SiS アプリを作成します。
Streamlit in Snowflake のアプリを実行
Streamlit in Snowflake アプリの編集画面で以下コードをコピー＆ペーストで貼り付けて完了です。
import os
import io
import pandas as pd
import streamlit as st
from snowflake.snowpark.context import get_active_session
# -------------------------------------
# Snowflakeセッションの取得
# -------------------------------------
session = get_active_session()
# -------------------------------------
# 定数・設定値
# -------------------------------------
# プレビュー可能なファイル拡張子
PREVIEWABLE_EXTENSIONS = ['.csv', '.txt', '.tsv']
# -------------------------------------
# ステージの存在チェックと作成処理
# -------------------------------------
def ensure_stage_exists(stage_name_no_at: str):
"""
指定されたステージが存在しない場合は作成する。既に存在する場合は何もしない。
"""
try:
# ステージが存在するかチェック
session.sql(f"DESC STAGE {stage_name_no_at}").collect()
except:
# 存在しない場合はステージ作成
try:
session.sql(f"""
CREATE STAGE {stage_name_no_at}
ENCRYPTION = (TYPE = 'SNOWFLAKE_SSE')
""").collect()
st.sidebar.success(f"ステージ @{stage_name_no_at} を作成しました。")
except Exception as e:
st.sidebar.error(f"ステージの作成に失敗しました: {str(e)}")
st.stop()
# -------------------------------------
# Streamlitアプリのメイン部分
# -------------------------------------
def main():
st.title("Snowflakeファイル管理アプリ")
# -------------------------
# ステージ設定
# -------------------------
st.sidebar.header("ステージ設定")
stage_name_no_at = st.sidebar.text_input(
"ステージ名を入力してください (例: MY_INT_STAGE)",
"MY_INT_STAGE"
)
stage_name = f"@{stage_name_no_at}"
# ステージが存在しない場合は作成
ensure_stage_exists(stage_name_no_at)
# -------------------------
# タブの作成
# -------------------------
tab_upload, tab_url, tab_download = st.tabs([
"ファイルアップロード",
"署名付きURL生成",
"ファイルダウンロード"
])
# -------------------------
# ファイルアップロードタブ
# -------------------------
with tab_upload:
st.header("ファイルアップロード")
st.write("ファイルをSnowflakeステージにアップロードします。")
uploaded_file = st.file_uploader("ファイルを選択してください")
if uploaded_file:
file_extension = os.path.splitext(uploaded_file.name)[1].lower()
try:
# BytesIOを使用してファイルストリームを作成し、アップロード
file_stream = io.BytesIO(uploaded_file.getvalue())
session.file.put_stream(
file_stream,
f"{stage_name}/{uploaded_file.name}",
auto_compress=False,
overwrite=True
)
st.success(f"ファイル '{uploaded_file.name}' のアップロードが完了しました！")
# アップロードファイルのプレビュー
if file_extension in PREVIEWABLE_EXTENSIONS:
try:
uploaded_file.seek(0)
if file_extension == '.csv':
try:
df_preview = pd.read_csv(uploaded_file)
except UnicodeDecodeError:
uploaded_file.seek(0)
df_preview = pd.read_csv(uploaded_file, encoding='shift-jis')
else: # .txt, .tsvなど
try:
df_preview = pd.read_csv(uploaded_file, sep='\t')
except UnicodeDecodeError:
uploaded_file.seek(0)
df_preview = pd.read_csv(uploaded_file, sep='\t', encoding='shift-jis')
st.write("アップロードされたデータのプレビュー:")
st.dataframe(df_preview.head())
except Exception as e:
st.warning(f"プレビューの表示中にエラーが発生しました: {str(e)}")
except Exception as e:
st.error(f"ファイルのアップロード中にエラーが発生しました: {str(e)}")
# -------------------------
# 署名付きURL生成タブ
# -------------------------
with tab_url:
st.header("署名付きURL生成")
st.write("ステージ上のファイルの署名付きURLを生成します。")
# ステージ上のファイル一覧を取得
stage_files = session.sql(f"LIST {stage_name}").collect()
if stage_files:
file_names = [
row['name'].split('/', 1)[1] if '/' in row['name'] else row['name']
for row in stage_files
]
with st.form("url_generation_form"):
selected_file = st.selectbox(
"URLを生成するファイルを選択してください",
file_names
)
expiration_days = st.slider(
"署名付きURLの有効期間 (日) を選択してください",
min_value=1,
max_value=7,
value=1,
help="1日から7日までの期間を選択できます"
)
submitted = st.form_submit_button("URLを生成")
if submitted:
try:
expiration_seconds = expiration_days * 24 * 60 * 60
url_statement = f"""
SELECT GET_PRESIGNED_URL(
'@{stage_name_no_at}',
'{selected_file}',
{expiration_seconds}
)
"""
result = session.sql(url_statement).collect()
signed_url = result[0][0]
st.success("URLの生成が完了しました！")
st.write(f"署名付きURL（{expiration_days}日間有効）:")
st.code(signed_url)
except Exception as e:
st.error(f"エラーが発生しました: {str(e)}")
else:
st.warning("ステージにファイルが存在しません。")
# -------------------------
# ファイルダウンロードタブ
# -------------------------
with tab_download:
st.header("ファイルダウンロード")
st.write("ステージ上のファイルをダウンロードします。")
# ステージ上のファイル一覧を取得
stage_files = session.sql(f"LIST {stage_name}").collect()
if stage_files:
file_names = [
row['name'].split('/', 1)[1] if '/' in row['name'] else row['name']
for row in stage_files
]
selected_file = st.selectbox(
"ダウンロードするファイルを選択してください",
file_names
)
if st.button("ダウンロード"):
try:
with session.file.get_stream(f"{stage_name}/{selected_file}") as file_stream:
file_content = file_stream.read()
st.download_button(
label="ファイルをダウンロード",
data=file_content,
file_name=selected_file,
mime="application/octet-stream"
)
except Exception as e:
st.error(f"エラーが発生しました: {str(e)}")
else:
st.warning("ステージにファイルが存在しません。")
# -------------------------------------
# アプリの起動
# -------------------------------------
if __name__ == "__main__":
main()
最後に
いかがでしたでしょうか？例えば手元の資材を誰かに共有する際に、メールでファイルを添付するのではなく、署名付き URL でサッと渡せると便利なのではないかと思います。st.file_uploader が Streamlit in Snowflake で使えるようになったことで更にデータアプリの実装の幅が増えますので、是非皆様も色々なアイディアを試してみていただければ嬉しいです！
宣伝
生成AI Conf 様の Webinar で登壇しました！
『生成AI時代を支えるプラットフォーム』というテーマの Webinar で NVIDIA 様、古巣の AWS 様と共に Snowflake 社員としてデータ*AI をテーマに LTをしました！以下が動画アーカイブとなりますので是非ご視聴いただければ幸いです！
SNOWFLAKE WORLD TOUR TOKYO のオンデマンド配信中！
Snowflake の最新情報を知ることができる大規模イベント『SNOWFLAKE WORLD TOUR TOKYO』が2024/9/11-12@ANAインターコンチネンタル東京で開催されました。
現在オンデマンド配信中ですので数々の最新のデータ活用事例をご覧ください。
また私が登壇させていただいた『今から知ろう！Snowflakeの基本概要』では、Snowflakeのコアの部分を30分で押さえられますので、Snowflake をイチから知りたい方、最新の Snowflake の特徴を知りたい方は是非ご視聴いただければ嬉しいですmm
X で Snowflake の What's new の配信してます
X で Snowflake の What's new の更新情報を配信しておりますので、是非お気軽にフォローしていただければ嬉しいです。
日本語版
Snowflake の What's New Bot (日本語版)
English Version
Snowflake What's New Bot (English Version)
変更履歴
(20250216) 新規投稿
(20250313) st.file_uploader の GA に伴い追記
Discussion