Streamlit in Snowflakeにコンテナランタイムがやってきた！

Posted at 2025-12-24

Streamlit in Snowflake Container runtimeとは

Streamlit in Snowflakeでコンテナランタイム(Container runtime)という新しい実行環境がリリースされました。
2025年12月時点ではまだPublic Previewです。

コンテナランタイムの使い方や注意事項は以下のブログが世界一分かりやすいのでチェックしておきましょう！

コンテナランタイムの利点は大きく3つだと考えています。

利用頻度が高いアプリのコストを大きく削減できる！
Streamlitの新しいバージョンを素早く導入できる！
パッケージの自由度が高い！

これまでのStreamlit in Snowflakeは、SnowflakeがSQLを実行するときに使う「仮想ウェアハウス」というコンピューティングリソースを使っていたのですが、これはStreamlitに必要な処理性能に対して割高感がありました。
一方、SnowflakeにはSnowPark Container Servicesというコンテナ基盤も提供されており、最小の仮想ウェアハウスと最小のコンテナ用コンピューティングリソースで比較すると1時間あたりのコストがなんと16分の1程度で済みます…！
ということで、軽量なアプリだったらこの最小のコンテナ用コンピューティングリソースを使えたらお得やん？ということで待望されていたのがコンテナランタイムです。

なお、Streamlitの内部的な組み込み方が従来の仮想ウェアハウスで動かす場合とコンテナランタイムで動かす場合で違うようです。
コンテナランタイムで動かすと設定項目が色々と違っていて、普通にコンテナで動いてそうー！と感じられますね…

コンテナランタイムで動かすまで

とりあえずコンテナ用のコンピュートリソースである、コンピュートプールを作成します。

-- Streamlit in Snowflake用のコンピュートプールを作成する
CREATE COMPUTE POOL STREAMLIT_COMPUTE_POOL
    MIN_NODES = 1
    MAX_NODES = 3
    INSTANCE_FAMILY = CPU_X64_XS;
DESCRIBE COMPUTE POOL streamlit_compute_pool;
SHOW SERVICES IN COMPUTE POOL streamlit_compute_pool;

AUTO_SUSPEND_SECSについて

以下のように自動サスペンドに関する設定を入れてもいいのですが、立ち上がりっぱなしになるため、あまり意味がありませんでした。

-- Streamlit in Snowflake用のコンピュートプールを作成する
CREATE COMPUTE POOL STREAMLIT_COMPUTE_POOL
    MIN_NODES = 1
    MAX_NODES = 3
    INITIALLY_SUSPENDED = TRUE
    AUTO_SUSPEND_SECS = 180
    INSTANCE_FAMILY = CPU_X64_XS;

また、ネットワークインテグレーションと呼ばれる、コンテナ内から特定のサイトにインターネットアクセスするための設定を行います。
これはPythonのパッケージリポジトリにアクセスするためです。
ネットワークルールを定義して、そのネットワークルールを参照する形でネットワークインテグレーションを設定するのですが、PyPIへのアクセスについてはSnowflakeに組み込まれているデフォルトのネットワークルールがあるので、それを使います。

-- PyPIにアクセスするためのネットワークインテグレーションを作成
CREATE OR REPLACE EXTERNAL ACCESS INTEGRATION PYPI_ACCESS_INTEGRATION
    ALLOWED_NETWORK_RULES = (snowflake.external_access.pypi_rule)
    ENABLED = TRUE;
DESCRIBE EXTERNAL ACCESS INTEGRATION PYPI_ACCESS_INTEGRATION;

この後、せっかくなので最新版のStreamlitを使おうと思ってstreamlit-nightlyのインストールを試したところ、pythonhosted.orgにアクセスできないというエラーが表示されたので、そちらにアクセスするための設定も行います。
おそらくどこかで私が設定ミスか操作ミスをしているせいなのですが、上述のネットワークインテグレーションが上手く動かないことがあったため、下記のようにPyPIへのアクセスも手動で設定しています。

-- Streamlit in Snowflakeのネットワークルールを作成
CREATE OR REPLACE NETWORK RULE PYTHONHOSTED_RULE
  MODE = EGRESS
  TYPE = HOST_PORT
  VALUE_LIST = ('*.pypi.org:0', '*.pythonhosted.org:0');

-- Streamlit in Snowflakeのネットワークインテグレーションを作成
CREATE OR REPLACE EXTERNAL ACCESS INTEGRATION PYTHONHOSTED_ACCESS_INTEGRATION
    ALLOWED_NETWORK_RULES = (PYTHONHOSTED_RULE)
    ENABLED = TRUE;

作成したネットワークインテグレーションは、SQLでCREATE STREAMLITしてアプリを作成するときに有効化してもいいですが、Snowsightの右上の三点メニューからApp settingsを開いて有効化することもできます。

アプリを作成するときに実行環境としてコンテナランタイムを選んで、上述のネットワークインテグレーションを有効化したら、後はPythonを書くだけ！

例えば以下のようにStreamlitの新しいバージョンを指定してあげれば…

pyproject.toml

[project]
name = "streamlit-app"
requires-python = ">= 3.11"
version = "0.0.1"
description = ""
dependencies = [
    "streamlit-nightly[snowflake]"
]

[tool.uv]
# earlier versions of numba require a python version < 3.11
constraint-dependencies = ["numba>=0.56.0"]

streamlit_app.py

import streamlit as st

st.title("コンテナランタイムで遊ぼう")
st.caption(f"Streamlit version: {st.__version__}")

prompt = st.chat_input("何か話してください。", accept_audio=True)
if prompt:
    if prompt.text:
        st.write(f"テキスト入力結果：{prompt}")
    elif prompt.audio:
        st.caption("音声入力結果：")
        st.audio(prompt.audio)

Streamlit 1.52.0でリリースされたchat_inputからの音声入力を使うことができました。

色んなパッケージも導入してみよう

従来のコンテナランタイムで動かすときはパッケージリポジトリがAnacondaリポジトリに限定されていました。それはそれで安心と信頼があるのですが、やっぱりそこにはないパッケージを使いたくなることもある…コンテナランタイムで動かす場合は、外部のパッケージリポジトリを利用することが可能です。
例えばノーコードのワークフローUIをStreamlit上で組めるBarfiを導入してみましょう！

まずはpyproject.tomlを更新します。

pyproject.toml

[project]
name = "streamlit-app"
requires-python = ">= 3.11"
version = "0.0.1"
description = ""
dependencies = [
    "streamlit[snowflake]",
    "barfi[streamlit]"
]

[tool.uv]
# earlier versions of numba require a python version < 3.11
constraint-dependencies = ["numba>=0.56.0"]

だいたいなんかおかしくなるので、手動でリブートしてあげます。

メインプログラムも修正してあげれば…

streamlit_app.py

import streamlit as st

st.title("コンテナランタイムで遊ぼう")
st.caption(f"Streamlit version: {st.__version__}")

# ここからはBarfiのチュートリアルコード
from barfi.flow import Block, SchemaManager, ComputeEngine
from barfi.flow.streamlit import st_flow

# "Number"ブロックの定義
number_block = Block(name="Number")
number_block.add_output(name="Output 1")
number_block.add_option(
    name="display-option", type="display", value="This is a Block with Number option."
)
number_block.add_option(name="number-block-option", type="number")
 
def number_block_func(self):
    number_value = self.get_option(name="number-block-option")
    self.set_interface(name="Output 1", value=number_value)
 
number_block.add_compute(number_block_func)

# "Result"ブロックの定義
result_block = Block(name="Result")
result_block.add_input(name="Input 1")
result_block.add_option(
    name="display-option", type="display", value="This is a Result Block."
)

def result_block_func(self):
    number_value = self.get_interface(name="Input 1")
    st.write(f"Resultブロックへの入力: {number_value}")
 
result_block.add_compute(result_block_func)

# NumberとResultが使えるフローを組み立てる
base_blocks=[number_block, result_block]
barfi_result = st_flow(base_blocks)
 
# フロースキーマをデバッグ出力する
if not barfi_result:
    st.stop()
with st.expander("フロースキーマ"):
    st.write(barfi_result)

# コンピュートエンジン(フロースキーマを実行するインスタンス)を生成
compute_engine = ComputeEngine(base_blocks)
 
# フロースキーマを実行
flow_schema = barfi_result.editor_schema
compute_engine.execute(flow_schema)

# 実行結果を出力する！
result_block = flow_schema.block(node_label="Result-1")
result_block_input_interface_value = result_block.get_interface("Input 1")
st.write(f"フロースキーマから得られた実行結果は{result_block_input_interface_value}です")

無事に動きました！

まとめ

現時点では、コンピュートプールが自動サスペンドしないなどの注意事項がありますが(最初に紹介したブログを参照)が色々ありますが、かなり将来性のあるアップデートだと感じました！
ぜひ遊んでみてください！

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up