カテゴリー: Snowflake

はじめに Streamlit in Snowflakeの開発を開始するには、Snowflakeアカウント、適切なIDE設定、ローカル開発環境の構築といった複数のステップが必要。この記事では、前提条件の確認、アプリケーション実装といった標準的なセットアップ手順をまとめる。 前提条件と必須の準備作業 Streamlit in Snowflakeの開発を始める前に、複数の前提条件を満たす必要がある。 前提条件の詳細： Snowflakeアカウントへのアクセス - 有効なSnowflakeアカウントと、CREATE APPLICATION PACKAGE 権限を持つロールが必須である。ロール設計を行い、この権限を付与したカスタムロールを使用する Pythonの開発環境 - Python 3.8以上がインストールされており、pipやcondaといったパッケージ管理ツールが動作する状態が前提である。Streamlit in SnowflakeはPython 3.10以上での動作を推奨している Snowparkライブラリ - ローカル開発環境にsnowpark、snowflake-snowpark-python といったパッケージをインストール済みであることが必須 Snowflake CLIツール - Snowflake提供の公式CLIツール（snow）をシステムに導入する必要がある。このツールを通じてSnowflakeを対話的に操作する 認証情報の管理 - ローカル開発では、Snowflakeへの接続情報をコードに埋め込まないことが重要である。環境変数、~/.snowsql/config ファイル、またはキーペア認証を使用して管理する。本番環境へのデプロイ時には、AWS Secrets Manager、Azure Key Vault、HashiCorp Vaultといった外部認証サービスの利用が推奨される IDE統合と開発環境の構築 Visual Studio Codeの統合により、ローカル開発フェーズ全体をエディタ内で完結させられる。Pythonコード編集、ローカルテスト実行（`streamlit run` での動作確認）、Snowflakeへのクエリ検証、デプロイまでの準備がVS Code内で実現される。一方、デプロイ後の本番環境ではSnowflakeウェブコンソール内でアプリケーションが動作する。公式のSnowflake拡張機能を利用することで、Snowflakeへの接続管理、SQL文の実行、デバッグが統一されたインターフェース内で実現される。 IDE統合のセットアップ手順： Visual Studio CodeにSnowflake拡張機能をインストールする。拡張機能マーケットプレイスから「Snowflake」を検索し、公式のSnowflake Inc.提供版をインストールする 拡張機能をインストール後、接続設定ファイル（通常は~/.snowsql/config）を確認し、接続情報が正確に記述されていることを検証する コマンドパレット（Ctrl+Shift+P または Cmd+Shift+P）からSnowflakeの接続を確立する。接続テストが成功することで、Snowflakeへの通信が確認される SQLエディタを開き、簡単なクエリ（例：SELECT CURRENT_USER()）を実行してSnowflakeとの疎通確認を行う IDE統合によって、ローカルでのPythonコード編集とSnowflakeのデータ参照が同一画面で実現され、開発の効率が劇的に向上する。 GitHub Codespacesでの開発も可能： ローカルマシンの環境管理を避けたい場合、GitHub Codespacesを使用した開発も実用的に使用できる。CodespacesにおいてSnowflake拡張機能、Snowflake CLI、Streamlit CLIがサポートされている。`streamlit run` コマンドで起動したアプリケーションはCodespaces内で自動的にポート転送され、ブラウザプレビューが利用可能である。環境構築を.devcontainer/devcontainer.jsonで定義すれば、チーム全体で統一された開発環境を即座に立ち上げられる。 ローカル開発環境のセットアップ ローカルマシンでStreamlit in Snowflakeアプリケーションを開発するには、複数のPythonパッケージが必要である。仮想環境の構築を通じて、プロジェクト固有の依存関係を隔離することが実務上の標準である。 開発環境の選択肢について： Docker環境を用いてローカル開発環境を構築することも技術的には可能だが、本番環境がSnowflake内の管理コンテナ上で実行されるため、ローカルのDocker環境と本番環境の構成を統一することはできない。開発環境をDockerで隔離したとしても、本番デプロイ時には別の実行環境へ移行するため、Docker化による環境共通化のメリットは限定的である。仮想環境による環境隔離で十分であり、Docker導入による複雑さの増加は費用対効果が低い。チーム規模が大きく、開発環境の統一が重要な場合のみDocker化を検討する価値がある。 # 仮想環境を作成 python3 -m venv streamlit_env # 仮想環境を有効化（macOS/Linux） source streamlit_env/bin/activate # 仮想環境を有効化（Windows） streamlit_envScriptsactivate # 必須パッケージをインストール pip install streamlit pip install snowflake-snowpark-python pip install snowflake-cli-labs # requirements.txtを作成し、プロジェクトの依存関係を記録 pip freeze > requirements.txt requirements.txtファイルの内容例： streamlit==1.28.0 snowflake-snowpark-python==1.10.0 snowflake-cli-labs==2.0.0 pandas==2.0.0 仮想環境の隔離により、異なるプロジェクト間での依存パッケージの競合を回避できる。これは本番環境へのデプロイ時にも重要であり、requirements.txtはアプリケーションと共にSnowflakeにアップロードされる。 ローカルでの初期テストと動作確認 ローカル開発環境が構築されたら、Streamlitが正常に動作するか確認する必要がある。最小限のアプリケーションコードでSnowflakeへの接続テストを行う。 # app.py import streamlit as st from snowflake.snowpark.context import get_active_session st.title(\"Streamlit in Snowflake - 初期テスト\") try: session = get_active_session() user = session.sql(\"SELECT CURRENT_USER()\").collect()[0][0] st.success(f\"Snowflakeへの接続成功。現在のユーザー: {user}\") except Exception as e: st.error(f\"接続エラー: {str(e)}\") # 簡単なデータクエリ if st.checkbox(\"テーブル一覧を表示\"): try: databases = session.sql(\"SHOW DATABASES\").collect() st.write(f\"利用可能なデータベース数: {len(databases)}\") except Exception as e: st.error(f\"クエリ実行エラー: {str(e)}\") このテストアプリケーションを実行する場合、ローカルではStreamlit CLIでの実行が可能である。 streamlit run app.py ただし、ローカルでの実行にはSnowflakeへの認証情報が必要である。環境変数で接続情報を提供する方法が一般的である。 必須の環境変数： SNOWFLAKE_ACCOUNT - Snowflakeアカウント識別子（例：xy12345.us-east-1） SNOWFLAKE_USER - ログインユーザー名 SNOWFLAKE_PASSWORD - ユーザーのパスワード（パスワード認証の場合） SNOWFLAKE_WAREHOUSE - クエリ実行用のウェアハウス名 SNOWFLAKE_DATABASE - デフォルトのデータベース名 SNOWFLAKE_SCHEMA - デフォルトのスキーマ名 キーペア認証を用いる場合は、SNOWFLAKE_PASSWORD の代わりに SNOWFLAKE_PRIVATE_KEY_PATH と SNOWFLAKE_PRIVATE_KEY_PASSPHRASE を設定する。環境変数の設定例： export SNOWFLAKE_ACCOUNT=\"xy12345.us-east-1\" export SNOWFLAKE_USER=\"developer_user\" export SNOWFLAKE_PASSWORD=\"your_secure_password\" export SNOWFLAKE_WAREHOUSE=\"dev_warehouse\" export SNOWFLAKE_DATABASE=\"analytics_db\" export SNOWFLAKE_SCHEMA=\"dev_schema\" # その後、streamlit run app.py を実行 streamlit run app.py 別の方法として、~/.snowsql/config ファイルに接続情報を記述し、Snowpark が自動的に読み込む設定も可能である。 初めてのアプリケーション実装 前提条件とローカル環境が整備されたら、Snowflakeアカウント内に実際のアプリケーションを作成する準備が整う。最小限のアプリケーションを実装し、Snowflakeへのデプロイが正常に機能することを確認する。 最小限のアプリケーション実装： # app.py import streamlit as st st.title(\"初めてのアプリケーション\") st.write(\"Hello World.\") このシンプルな実装で、Streamlit in Snowflakeへのデプロイが正常に完了し、本番環境でUIが表示されることを確認できる。Snowflake側でアプリケーション作成用のステージとメタデータを準備する必要がある。 -- Snowflakeで実行：アプリケーション用のステージを作成 CREATE STAGE IF NOT EXISTS app_stage; -- アプリケーション設定ファイルを準備 -- manifest.ymlを作成してステージにアップロード Snowflake CLIを使用して、ローカルのアプリケーションコードをSnowflakeにデプロイする。 # Snowflake CLIでプロジェクトを初期化 snow project init # ローカルの開発コードをSnowflakeにデプロイ snow app deploy デプロイ後の検証： Snowflakeウェブコンソールにログインし、アプリケーション一覧から新規作成したアプリケーションが表示されていることを確認する アプリケーションをクリックして開き、UIが正常に表示され、Snowflakeへのクエリが実行される状態を確認する デプロイ後の最初の実行はコールドスタートのため、数秒の遅延が発生するが、以後のアクセスは高速化される 以下はエラーハンドリングを組み込んだ実装例。Snowflake環境において発生したネットワークエラー、タイムアウト、権限不足といった例外を補足し表示してみた。 # エラーハンドリングを含む実装例 import streamlit as st from snowflake.snowpark.context import get_active_session st.set_page_config(page_title=\"データダッシュボード\", layout=\"wide\") try: session = get_active_session() st.header(\"データ参照アプリケーション\") # ユーザー情報の取得 current_user = session.sql(\"SELECT CURRENT_USER()\").collect()[0][0] st.sidebar.write(f\"ユーザー: {current_user}\") # データベース選択 db_list = session.sql(\"SHOW DATABASES\").collect() databases = [row[1] for row in db_list] selected_db = st.selectbox(\"データベースを選択\", databases) st.success(f\"接続完了\") except Exception as e: st.error(f\"エラーが発生しました: {type(e).__name__}\") st.info(\"管理者に連絡してください\") まとめ 本記事では、前提条件の確認、IDE統合（Visual Studio Code、Snowflake拡張機能のセットアップ）、GitHub Codespacesでの開発環境構築の検討、ローカル開発環境の準備（仮想環境、パッケージインストール、Docker化の考慮を含む）について言及した。また、ローカルテスト実行時の環境変数設定方法についてまとめた。最後に、最小限のアプリケーションを実装し、Snowflakeへデプロイ後、動作確認を行なった。

はじめに Streamlit in Snowflakeで本番環境のアプリケーションを構築する際、実行環境とセッション管理の仕組みを理解することは必須である。標準的なStreamlitとは異なり、Snowflake統合版はSnowflakeの管理するコンテナ内で実行され、アプリケーションのライフサイクル、パフォーマンス特性、状態管理が大きく異なる。本稿では、この実行モデルの核心部分に焦点を当て、本番環境での実装判断に必要な知識を整理する。標準的なStreamlitの開発経験がある技術者であっても、Snowflake統合版の独特なアーキテクチャを把握することで、より堅牢で効率的なアプリケーション設計が可能となる。 Snowflakeの管理するコンテナ内での実行 Streamlit in Snowflakeのアプリケーションは、Snowflakeのアカウント内で管理された隔離されたコンテナプロセス上で実行される。ローカルマシンのPythonプロセスのように直接制御することはなく、Snowflakeのインフラストラクチャが実行環境全体を統制する。 実行環境の核心的な特性： 各アプリケーションはSnowflakeのアカウント領域内で独立した仮想環境として分離されており、他のテナントや他のアプリケーションとの干渉を受けない アプリケーションの起動、実行、終了はSnowflakeの制御下にあり、ユーザーのアクセスパターンに応じた動的なスケーリングが自動的に実行される Pythonランタイムは事前にコンテナ内にプリロードされており、ユーザーがアプリケーションにアクセスした時点でコードの実行が即座に開始される コンテナはステートレスな設計であり、複数のユーザーセッション間でローカルのファイルシステム上の状態は保持されない メモリ、CPU、ネットワーク帯域幅などのリソースは制限されており、無限に大規模なデータセットをメモリに展開することはできない この設計により、スケーラビリティと管理負荷の削減が実現される。開発者はインフラストラクチャの保守運用から解放され、アプリケーション本体の開発に集中できる。一方で、アプリケーション開発者は「各セッションは独立している」「ローカル状態は永続しない」という前提でコーディングする必要があり、この認識がなければ本番環境で予期しない動作が発生する可能性がある。 ExecutionContextとSnowflakeのセッション情報へのアクセス Streamlit in Snowflakeで最も重要な概念がExecutionContextである。これはSnowflakeのセッション情報とアプリケーション実行の状態を統合したオブジェクトであり、アプリケーションコード内から直接アクセスすることが可能である。 ExecutionContextを通じて、認証済みユーザーの識別子、割り当てられたウェアハウス、セッションのロール情報、現在のデータベースとスキーマといった情報が取得できる。これらの情報はSnowflakeの権限管理体系と一体化しており、アプリケーションが実行するすべてのSQL文はこのコンテキストの権限に基づいて検証される。 from snowflake.snowpark.context import get_active_session session = get_active_session() # 現在のユーザーを取得 current_user = session.sql(\"SELECT CURRENT_USER()\").collect()[0][0] # 割り当てられたウェアハウスを確認 current_warehouse = session.sql(\"SELECT CURRENT_WAREHOUSE()\").collect()[0][0] # 現在のロール情報 current_role = session.sql(\"SELECT CURRENT_ROLE()\").collect()[0][0] # アプリケーション領域のスキーマを取得 current_schema = session.sql(\"SELECT CURRENT_SCHEMA()\").collect()[0][0] ExecutionContextから取得可能な情報の実用的な用途： 認証済みユーザーID：このユーザーが属するテナント、部門、権限レベルをデータベースから検索し、表示内容を動的に制御する基盤となる 割り当てられたウェアハウス：クエリの実行リソースがどのウェアハウスに割り当てられるかを把握し、多くの重い処理が実行される時間帯を避けるといった最適化判断に活用される セッションのロール情報：ロールベースアクセス制御の実装において、現在のユーザーが実行可能な操作を制限する際に利用される 現在のデータベースおよびスキーマ：アプリケーションが参照するテーブルやストアドプロシージャの名前空間を把握し、正確なクエリを構築する際に用いられる ExecutionContextはSnowflakeの行レベルセキュリティ（RLS）および動的データマスキング（DDM）と組み合わせることで、マルチテナント環境でのデータ分離を実装できる。ユーザーが属するテナント情報をExecutionContextから抽出し、その情報をSQLクエリに動的にフィルタリング条件として付与するパターンが一般的である。 セッション状態の管理と永続化戦略 Streamlit in Snowflakeでは、標準的なStreamlitの`st.session_state`メカニズムが使用される。ただし、その永続性と可用性については、通常のStreamlitと異なる考慮が必要である。 セッション状態の保持期間と動作： ユーザーがブラウザを閉じるまで、またはセッションのタイムアウト（デフォルト約60分）が発生するまで、`st.session_state`に格納されたPythonオブジェクトは保持される セッション終了後、メモリ上の状態は完全に消失し、その後のユーザーアクセスでは初期化された状態から再出発する 複数のユーザーセッションが並行して実行される場合、各セッションのメモリ空間は完全に独立しており、相互干渉は発生しない 分散環境ではコンテナのリバランシングが発生する可能性があり、メモリ内状態への依存度が高いと予期しない状態喪失が発生する危険性がある セッション状態を効果的に使用するパターンとしては、ユーザーの入力フォーム状態、フィルタ条件、ページネーション状態、一時的なキャッシュなど、セッション内での短期的な状態に限定することが推奨される。 import streamlit as st from snowflake.snowpark.context import get_active_session session = get_active_session() # セッション状態で一時的なUIフィルタを保持 if \'selected_date_range\' not in st.session_state: st.session_state.selected_date_range = (None, None) if \'filter_status\' not in st.session_state: st.session_state.filter_status = \'all\' # ユーザーインタラクションでセッション状態を更新 date_range = st.date_input(\"期間を選択\", st.session_state.selected_date_range) st.session_state.selected_date_range = date_range # 永続化が必要な設定はSnowflakeテーブルに明示的に保存 if st.button(\'設定を保存\'): current_user = session.sql(\"SELECT CURRENT_USER()\").collect()[0][0] session.sql(f\"\"\" UPDATE user_preferences SET ui_settings = parse_json(?) WHERE user_id = ? \"\"\", params=[str(st.session_state.filter_status), current_user]).collect() st.success(\"設定を保存しました\") セッション終了後も保持する必要があるデータ（ユーザー設定、保存された検索条件、レポート結果など）は、Snowflakeのテーブルに明示的に書き込む必要がある。この明確な分離により、アプリケーションの動作が予測可能になり、バグの温床となる隠れた状態共有が回避される。 パフォーマンス特性とコールドスタート最適化 Streamlit in Snowflakeのパフォーマンス特性は、コンテナの起動時間、リソースの割り当て、クエリの実行効率によって大きく影響を受ける。 パフォーマンスに関わる重要な指標： 初期化時間：ユーザーがアプリケーションに初めてアクセスする際、Snowflakeがコンテナを起動し、Pythonランタイムを初期化するまでに3秒から10秒程度要する場合がある。これをコールドスタートと呼ぶ SQLクエリ実行時間：SQLクエリの実行時間は主にSnowflakeのクエリプランニングと分散処理の時間に依存し、ネットワークレイテンシは相対的に最小限である メモリ制限：各コンテナプロセスのメモリは制限されており、gigabyte単位の大規模なデータセットを一度にメモリにロードすることは技術的に不可能である リソース競合：同一のウェアハウス上で複数のアプリケーションやクエリが並行実行される場合、リソース争奪による性能低下が発生する可能性がある キャッシュ効果：頻繁にアクセスされるテーブルやクエリ結果はSnowflakeの内部キャッシュに保持され、2度目以降のアクセスは高速化される 本番環境ではコールドスタート対策として、アプリケーション初期化時の処理を最小化し、必要なデータは遅延読み込みするパターンが採用される。また、複雑な分析処理やデータ変換はSnowflakeのストアドプロシージャに委譲し、アプリケーション層では結果の表示と対話的なUIの提供に専念することが効率的である。 ウェアハウスとリソース割り当ての考慮 Streamlit in Snowflakeのアプリケーションが実行するすべてのSQL文は、指定されたウェアハウスのコンピュート能力を消費する。ウェアハウスの選択は、クエリの実行速度、同時実行可能なセッション数、運用コストに大きな影響を与える。 ウェアハウス選択の実務的考慮： 小規模なウェアハウス（XSMALL、SMALL）はコストが低く、軽量なクエリや低アクセス頻度のアプリケーションに適しており、一方で大量のユーザーからの並行アクセスには不向きである 大規模なウェアハウス（LARGE、XLARGE以上）は並行クエリ処理の能力が高く、多くのユーザーからの同時アクセスに対応できるが、アイドル状態であってもコストが発生する オートスケール機能を有効にすることで、負荷に応じたウェアハウスの自動拡張が可能になり、ピーク時の対応と非ピーク時のコスト削減を両立させられる 複数のアプリケーションが同一ウェアハウスを共有する場合、負荷分散戦略を立案し、一つのアプリケーションの過度なリソース消費が他のアプリケーションに悪影響を与えないようにする必要がある リソースの効率的な利用と高いパフォーマンスの両立は、ウェアハウスのサイズ選択、クエリの最適化、適切なキャッシング戦略によって初めて実現される。 まとめ Streamlit in Snowflakeは、Snowflakeの管理するマネージドコンテナ環境内で動作し、ExecutionContextを通じてSnowflake側のセッション情報に直接アクセスできるアーキテクチャである。セッション状態は短期的なUI状態の保持に限定し、永続化が必要なデータはSnowflakeテーブルに委譲することが設計の原則である。また、コールドスタートやリソース共有の課題を念頭に置いて、初期化処理の最小化とクエリの最適化によるパフォーマンス改善アプローチを検討する必要がある。これらの理解があれば、本番環境での実装判断が格段に容易になり、堅牢で拡張性の高いアプリケーション設計が可能となる。

何周遅れか分からないが、Snowflake MCPサーバを試してみたのでアウトプットしてみる。 AI AgentはClaude Code。MCPの構築と接続設定自体をClaude Codeで行なった。 この記事で使用したMCPサーバは以下。いわゆる野良MCPサーバ。 [clink implicit=\"false\" url=\"https://github.com/isaacwasserman/mcp-snowflake-server\" imgurl=\"https://camo.githubusercontent.com/bdcfca988b369e51051c3201cedfc429354b0801a0c5d88aa3eb00ae37e7188b/68747470733a2f2f6d736565702e6e65742f70722f69736161637761737365726d616e2d6d63702d736e6f77666c616b652d7365727665722d62616467652e706e67\" title=\"Snowflake MCP Server\" excerpt=\"A Model Context Protocol (MCP) server implementation that provides database interaction with Snowflake. This server enables running SQL queries via tools and exposes data insights and schema context as resources.\"] [arst_toc tag=\"h4\"] 前提となる環境 Macにnode、uv、Claude Codeを導入済み。 # 諸々のバージョンは以下 $ sw_vers ProductName: macOS ProductVersion: 15.6 BuildVersion: 24G84 # nodeは導入済み $ node --version v24.4.1 # uvは導入済み $ uv --version 0.8.13 (ede75fe62 2025-08-21)0.8.13 (ede75fe62 2025-08-21) # Claude Codeは導入済み $ claude --version 1.0.89 (Claude Code) # 検証用ディレクトリの作成と移動。以降ここで検証を実施。 $ mkdir snowflake-mcp-server && cd $_ 環境構築 プロンプトとその回答は省略する。要件を伝え環境構築を実施してもらった。 Claudeに対するMCPの設定方法としてグローバルに設定する方法と、 プロジェクト個別に設定する方法があるが、当初グローバル設定の方法を案内された。 2回目のプロンプトでプロジェクト個別設定に変更してもらった。 最終的なファイル構成は以下の通り。 $ tree . . ├── mcp-snowflake-server │   ├── Dockerfile │   ├── example_connections.toml │   ├── LICENSE │   ├── pyproject.toml │   ├── README.md │   ├── runtime_config.json │   ├── src │   │   └── mcp_snowflake_server │   │   ├── __init__.py │   │   ├── db_client.py │   │   ├── serialization.py │   │   ├── server.py │   │   └── write_detector.py │   └── uv.lock ├── PROJECT_SPECIFIC_SETUP.md ├── SETUP.md ├── snowflake_connections.toml └── test_server.py snowflake_connections.toml に Snowflakeの接続設定を書くだけで終了。 試してみた claudeを起動すると .mcp.json が認識された。 $ claude ╭─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮ │ │ │ New MCP server found in .mcp.json: snowflake │ │ │ │ MCP servers may execute code or access system resources. All tool calls require approval. Learn more in the MCP documentation │ │ (​https://docs.anthropic.com/s/claude-code-mcp​). │ │ │ │ ❯ 1. Use this and all future MCP servers in this project │ │ 2. Use this MCP server │ │ 3. Continue without using this MCP server │ │ │ ╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯ > Snowflakeのデータベース `sample_db` におけるスキーマの一覧を教えてください。 ⏺ snowflake - list_schemas (MCP)(database: \"sample_db\") ⎿  type: data data_id: e40f8385-e3c0-4848-aa36-c1b8fada788f database: sample_db … +8 lines (ctrl+r to expand) ⎿  [Resource from snowflake at data://e40f8385-e3c0-4848-aa36-c1b8fada788f] { \"type\": \"data\", \"data_id\": \"e40f8385-e3c0-4848-aa36-c1b8fada788f\", … +25 lines (ctrl+r to expand) ⏺ sample_db データベースには以下の7つのスキーマがあります： - CLEANSED - INTERMEDIATE - PUBLIC - SAMPLE_SCHEMA > cleansedスキーマ内のテーブルの一覧を教えてください。 ⏺ CLEANSED スキーマには以下の19のテーブルがあります： ... メタ情報(description) dbtでモデルを作る際にdescriptionを付与することで、テーブル、カラムにdescriptionを付与できる。 その先、セマンティックレイヤーの存在がチラつくが、descriptionだけでもそこそこいける。 まとめ Claude Codeにローカル起動型(Dockerfile)のSnowflake MCPサーバを導入してみた。 その際、Claude Code自身にMCPサーバの導入を依頼し、ほぼノーコードで導入できることを確認した。 テーブル、カラムにdescriptionを付与しておくと、MCPサーバを通じて取得・消費される。 クエリによって得られる知識がClaudeに蓄積され、知識の文脈で会話ができるようになることを確認した。

シンプルな Multi-Container App を動かしている以下の記事にインスパイアされてみました。 以下の記事では、Docker networkを前提にフロントがサーバの名前解決を行っています。 これをデプロイすると、ブラウザで動くフロントコードがサーバの名前を解決できません(SPCS無関係)。 リバースプロキシを挟んでプライベートなダウンストリームにAPIを配置する方法が良さそうです。 今回の記事はSPCSの動作確認をすることが目的なので凝ったことはせず、 ViteをそのままデプロイしてProxyで解決してみたのでご紹介します。 [clink implicit=\"false\" url=\"https://medium.com/@maseedilyas9848/snowflake-container-mastery-step-by-step-deployment-of-your-multi-container-app-with-snowpark-211682514851\" imgurl=\"https://miro.medium.com/v2/resize:fit:1400/format:webp/1*t7s-Rl6F4BBV-yYs-ovODQ.png\" title=\"Snowflake Container Mastery: Step-by-Step Deployment of Your Multi-Container App with Snowpark Container Services\" excerpt=\"The buzz around town is all about Snowflake’s latest product feature, “Snowpark Container Services” and the excitement is real. Now, with the feature hitting public preview in various AWS regions, this blog dives into the nitty-gritty of what container services bring to the table. Join me as we explore what makes this feature tick and unravel the steps to deploy a multi-container app within Snowflake. Let’s break it down!\"] [arst_toc tag=\"h4\"] SPCSのアーキテクチャ Image Registryはイメージリポジトリです。AWSのECR、AzureのACRと似た感じで操作できます。 Container Serviceはデプロイメントの単位で、1個以上のコンテナをデプロイできます。 Compute Poolは計算資源で、複数のContainer Serviceが共有します。 Serviceは基本的にはPrivateなリソースとなりますが、SpecでEndpointsを定義することで、 Publicリソースとすることができます。自動的に80にmapされます。 Serviceには以下のフォーマットでDNS名が付きます。 Service同士は以下のDNS使ってプライベート通信できます。 同一DB、Schema内に作成したServiceは先頭のServiceNameが異なるだけとなりますが、 その場合に限り、ServiceNameだけで互いの名前解決ができます。 Service間連携については、公式で\"Service-to-Service\"というパターンで紹介されています。 <Service Name>-<Schema Name>-<DB Name>.snowflakecomputing.internal 1つのService内に複数のコンテナを配置することができます。 同一Service内で各コンテナ内の通信したいと考えたのですが、方法を見つけることができませんでした。 (出来ないはずはなさそうで方法はあるのかもしれません..) 今回作るもの フロント側、サーバ側の2つのコンテナを、それぞれ別々のServiceとしてデプロイします。 フロント側をPublic、サーバ側はPrivateとします。概要は以下の通りです。 フロント側 Vite React/TypeScript ReactからのAPIリクエストは一旦ViteのProxyで受けて、ProxyからAPIに流す サーバ側にGETリクエストして応答を表示するだけ サーバ側 uvicorn FastAPI poetry GETリクエストを受けて\"Hello World\"をJSONで返すだけ 思いっきり開発用な感じですが、SPCSの動作確認が目的ですのでこれでいきます。 ローカルで動作確認をして、SPCSにデプロイします。ファイル構成は以下の通りです。 $ tree . -I node_modules -I __pycache__ . ├── api │   ├── Dockerfile │   ├── api.py │   ├── app.py │   ├── main.py │   └── pyproject.toml ├── compose.yml └── front ├── Dockerfile ├── entry.sh ├── index.html ├── package-lock.json ├── package.json ├── src │   └── hello.tsx ├── tsconfig.json ├── tsconfig.node.json └── vite.config.ts compose.yml サーバ側、フロント側の2つのコンテナが、サーバ側->フロント側の順に起動するように書きます。 それぞれ、8080、5173 で待つようにします。 services: api: build: api volumes: - ./api:/app ports: - 8080:8080 front: build: front volumes: - ./front:/app ports: - 5173:5173 depends_on: - api サーバ側 Snowflakeの公式のチュートリアルだとFlaskが使われています。 今回は、最近使い始めたFastAPIを使ってAPIサーバを立ててみようと思います。 FlaskとFastAPIの比較はこちらが詳しいです。 FastAPIの特徴はPydanticによるデータ検証とAsyncI/O。TypeScriptのように型チェックできます。 パッケージマネージャにはpoetryを使います。デファクトが無いPythonのパッケージ管理界隈で npmやcomposer的な使い勝手が提供されます。 FastAPIの公式では、Python用Webサーバのuvicornを使ってホストされています。 uvicornでFastAPIを動かすコンテナを1個立てていきます。 Dockerfile 最新のPythonのイメージにpoetryをインストールします。 公式がインストーラを配布していて公式の手順通りに叩けばインストールできます。 公式のガイドの通り、POETRY_HOME=/etc/poetry とします。 Installation with the official installer FROM python:3.12 # 公式の通り /etc/poetry にインストールする ENV POETRY_HOME=/etc/poetry RUN curl -sSL https://install.python-poetry.org | python - ENV PATH $POETRY_HOME/bin:$PATH WORKDIR /app COPY . . RUN poetry install CMD [\"python\",\"main.py\"] pyproject.toml バニラのPythonとpyproject.tomlだけで依存関係を考慮したパッケージ管理が出来ますが、 要はnpmやbundle,composer的な使い勝手に寄せたパッケージ管理に対する需要があります。 poetry用の依存関係を書いていきます。fastapi、uvicornの現在(2/16)の最新を指定します。 [tool.poetry] name = \"test\" [tool.poetry.dependencies] python = \"^3.12\" fastapi = \"^0.109.2\" uvicorn = \"^0.27.1\" api.py [GET] /hello に対して Hello World 的な JSON を返す FastAPI の Hello Worldです。 後のapp.pyで FastAPIインスタンスに紐付けます。app.pyと分離することでロジックを分離できます。 from fastapi import APIRouter router = APIRouter() @router.get(\"/hello\") async def hoge(): return {\"result\":\"Hello World\"} app.py FastAPIのHello Worldコードです。Dockerfileから開始します。 from api import router as api_router from fastapi import FastAPI app = FastAPI() app.include_router(api_router, prefix=\"/api\") main.py pythonコードからuvicornを起動します。uvicorn.runの公式の仕様はこちら。 第1引数の\"app:app\"は\"app\"モジュールの中の\"app\"オブジェクトという表記です。 app.pyに記述したFastAPI()のインスタンスappを指します。stringで渡す必要があります。 hostは\"0.0.0.0\"を指定します。なぜ\"127.0.0.1\"でないのかはこちらが参考になります。 今回はPort=8080で起動します。reload=Trueとすると、HotReload機能が有効になります。便利。 import uvicorn if __name__ == \"__main__\": uvicorn.run( \"app:app\", host=\"0.0.0.0\", port=8080, reload=True, ) 起動してみる docker compose up して http://localhost:8080/docs を開くと以下が表示されます。 ちゃんとJSONでHello Worldが戻りました。 フロント側 VueとReactの開発用に使われるローカル開発用のサーバ Vite をホストするコンテナを立てます。 Viteは Vue.jsの開発者Evan You氏が開発したJavaScript/TypeScriptで、ヴィートと読み、 フランス語で\"素早い\"という意味だそう。(webpackのように)リソースバンドルが不要で起動が速い。 (Laravelも9.xでwebpackを捨ててViteになってた..) 素の状態でTypeScriptを扱えるため、すぐにTypeScriptを書き始められる特徴があります。 Dockerfile nodeのrelease scheduleはこちら。 2/17のnodeのActiveLTSのMajor Versionは20で、2026-04-30がEnd of lifeとなっています。 これを使いたいので node:20 を指定します。 FROM node:20 WORKDIR /app COPY . . RUN npm install ENTRYPOINT [ \"./entry.sh\" ] entry.sh Dockerfile の ENTRYPOINT で npm run dev するだけのshです。 #!/bin/bash npm run dev package.json npm create vite@latest で Prjディレクトリ内に様々なファイルが作られます。 package.jsonも作られます。 Hello World で必要なもの以外を削ってみました。 npm install後、npm run devで viteを実行します。 TS用のconfigは別です。 他に生成されるpackage-lock.jsonが必要ですが省略します。 { \"name\": \"front\", \"private\": true, \"version\": \"0.0.0\", \"scripts\": { \"dev\": \"vite\" }, \"dependencies\": { \"react\": \"^18.2.0\", \"react-dom\": \"^18.2.0\" }, \"devDependencies\": { \"@vitejs/plugin-react\": \"^4.2.1\" } } tsconfig.json viteのPrj生成で自動生成されるTS用のconfigファイルです。 手をつけずに配置します。 { \"compilerOptions\": { \"target\": \"ES2021\", \"useDefineForClassFields\": true, \"lib\": [\"ES2021\", \"DOM\", \"DOM.Iterable\"], \"module\": \"ESNext\", \"skipLibCheck\": true, /* Bundler mode */ \"moduleResolution\": \"bundler\", \"allowImportingTsExtensions\": true, \"resolveJsonModule\": true, \"isolatedModules\": true, \"noEmit\": true, \"jsx\": \"react-jsx\", /* Linting */ \"strict\": true, \"noUnusedLocals\": true, \"noUnusedParameters\": true, \"noFallthroughCasesInSwitch\": true }, \"include\": [\"src\"], \"references\": [{ \"path\": \"./tsconfig.node.json\" }] } tsconfig.node.json viteのPrj生成で自動生成されるTS用のconfigファイルです。 手をつけずに配置します。 { \"compilerOptions\": { \"composite\": true, \"skipLibCheck\": true, \"module\": \"ESNext\", \"moduleResolution\": \"bundler\", \"allowSyntheticDefaultImports\": true }, \"include\": [\"vite.config.ts\"] } vite.config.ts viteのPrj生成で自動生成されるvite用のconfigファイルです。 設定ファイルが.tsなところが凄いです。普通にimport文を書けます。 上のuvicornの起動で127.0.0.1ではなく0.0.0.0を指定したのと同様に、 viteも127.0.0.1ではなく0.0.0.0で待たせる必要があります。 serverオプションのhostにtrueを設定すると、0.0.0.0となります。公式 FastAPIの同一パスと対応するProxyを設定します。 以下で、server.proxy.api.target は SPCS上のAPIコンテナのPrivateエンドポイント を表します。 DNS名はサービス単位で作られます。本来長いFQDNを指定する必要がありますが、 同一スキーマに作られたサービスに限り、サービス名だけで解決できるようです。 DNS名はアンダースコア(_)がハイフン(-)に置き換わります。6時間くらいハマりました.. 後で ikuty_api_service サービスを作りますが、ikuty-api-serviceを 使います。 詳細は以下を参照してください。 Service-to-service communications import { defineConfig } from \'vite\' import react from \'@vitejs/plugin-react\' export default defineConfig({ plugins: [react()], server: { host: true, proxy: { \"/api\": { target: `http://ikuty-api-service:8080/`, changeOrigin: true } } }, }) index.html Reactのコンポーネントを表示するガワとなるhtmlです。 <!DOCTYPE html> <html> <head> <meta charset=\"UTF-8\" /> <meta name=\"viewport\" content=\"width=device-width, initial-scale=1.0\" /> <script type=\"module\" src=\"/src/hello.tsx\" defer></script> </head> <body> <div id=\"root\">Waiting...</div> </body> </html> src/hello.tsx ようやくHello WorldするReactコンポーネントの本体です。 画面にはvalueというStateを表示しています。APIのURLは上記の通りproxyとします。 今回作成した/api/hello APIの応答を受けた後、setValueによりStateを更新します。 import React from \'react\' import { useState } from \'react\' import ReactDOM from \'react-dom/client\' const App = () => { const [value, setValue] = useState(\'\') const url = \'/api/hello\' fetch(url,{}) .then(res=>res.json()) .then(data=>setValue(data[\'result\'])) return ( {value},{} ) } ReactDOM.createRoot(document.getElementById(\'root\')!).render( ) 起動してみる docker compose up すると、ほとんど一瞬でviteが起動します。 http://localhost:5173 を開きます。 Waiting...という表示が一瞬で Hello World に書き変わります。 ロールの作成 SPCSの各リソースの作成に必要な権限はこちらにあります。 ゴリ押ししただけなので間違っている可能性大です.. 行ったり来たりしたので足りないものがあるかもしれません。 use role ACCOUNTADMIN; CREATE ROLE IKUTY_CONTAINER_USER_ROLE; GRANT ROLE IKUTY_CONTAINER_USER_ROLE TO ROLE ACCOUNTADMIN; GRANT USAGE ON DATABASE IKUTY_DB TO ROLE IKUTY_CONTAINER_USER_ROLE; GRANT USAGE ON SCHEMA IKUTY_DB.PUBLIC TO ROLE IKUTY_CONTAINER_USER_ROLE; GRANT CREATE IMAGE REPOSITORY ON SCHEMA T_IKUTA_DB.PUBLIC TO ROLE IKUTY_CONTAINER_USER_ROLE; -- CREATE SERVICEに必要な権限 -- https://docs.snowflake.com/en/sql-reference/sql/create-service#access-control-requirements GRANT USAGE ON DATABASE IKUTY_DB TO ROLE IKUTY_CONTAINER_USER_ROLE; GRANT USAGE ON SCHEMA IKUTY_DB.PUBLIC TO ROLE IKUTY_CONTAINER_USER_ROLE; GRANT CREATE COMPUTE POOL ON ACCOUNT TO ROLE IKUTY_CONTAINER_USER_ROLE; GRANT CREATE IMAGE REPOSITORY ON SCHEMA IKUTY_DB.PUBLIC TO ROLE IKUTY_CONTAINER_USER_ROLE; GRANT CREATE SERVICE ON SCHEMA IKUTY_DB.PUBLIC TO ROLE IKUTY_CONTAINER_USER_ROLE; GRANT USAGE ON COMPUTE POOL IKUTY_SCS_POOL TO ROLE IKUTY_CONTAINER_USER_ROLE; GRANT BIND SERVICE ENDPOINT ON ACCOUNT TO ROLE IKUTY_CONTAINER_USER_ROLE; GRANT IMPORTED PRIVILEGES ON DATABASE snowflake TO ROLE IKUTY_CONTAINER_USER_ROLE; -- GRANT READ ON STAGE IKUTY_SCS_STAGE TO ROLE IKUTY_CONTAINER_USER_ROLE; GRANT READ ON IMAGE REPOSITORY IKUTY_SCS_REPOSITORY TO ROLE IKUTY_CONTAINER_USER_ROLE; GRANT BIND SERVICE ENDPOINT ON ACCOUNT TO ROLE IKUTY_CONTAINER_USER_ROLE; Image Repositoryの作成 SPCSで使用するイメージを配置するリポジトリを作成します。 AWSのECR、AzureのACR的に、dockerコマンドから透過的にpushできるようです。 公式は以下。 CREATE IMAGE REPOSITORY USE ROLE IKUTY_CONTAINER_USER_ROLE; CREATE OR REPLACE IMAGE REPOSITORY IKUTY_SCS_REPOSITORY; SHOW IMAGE REPOSITORIES; SHOW IMAGEを叩くと repository_url が返ってます。 Image Repositoryにプッシュする 作成したImage Repositoryにローカルで作成したイメージをpushしていきます。 pushは指定されたタグを送信するという仕様のため、docker tagコマンドでイメージにタグを付けます。 docker tagの仕様はこちら。 ローカルで以下を行います。(サニタイズのため分かりづらいですが補完してください..) # タグをつける # docker tag $ docker tag app_front:latest /app_front:scs $ docker tag app_api:latest /app_api:scs # Snowflake Image Repositoryにログインする $ docker login -u Login Succeeded # イメージをpushする $ docker push /app_front:scs ... $ docker push /app_api:scs ... Compute Poolを作成する Compute Poolを作成します。 CREATE COMPUTE POOL CREATE COMPUTE POOL ikuty_scs_pool MIN_NODES = 1 MAX_NODES = 1 INSTANCE_FAMILY = CPU_X64_XS AUTO_RESUME = TRUE INITIALLY_SUSPENDED = FALSE AUTO_SUSPEND_SECS = 3600 ; 以下でCREATEしたCompute poolをDESCRIBEできます。 DESCRIBE COMPUTE POOL 自分の環境だと、CREATE COMPUTE POOLしてから15分ほどステータスがSTARTINGでした。 15分ぐらいして叩くとステータスがACTIVEに変わりました。(結構かかるイメージ) 以下、公式の実行例です。 DESCRIBE ikuty_scs_pool +-----------------------+--------+-----------+-----------+-----------------+--------------+----------+-------------------+-------------+--------------+------------+-------------------------------+-------------------------------+-------------------------------+--------------+---------+ | name | state | min_nodes | max_nodes | instance_family | num_services | num_jobs | auto_suspend_secs | auto_resume | active_nodes | idle_nodes | created_on | resumed_on | updated_on | owner | comment | |-----------------------+--------+-----------+-----------+-----------------+--------------+----------+-------------------+-------------+--------------+------------+-------------------------------+-------------------------------+-------------------------------+--------------+---------| | IKUTY_SCS_POOL | ACTIVE | 1 | 1 | CPU_X64_XS | 1 | 0 | 0 | false | 1 | 0 | 2023-05-01 11:42:20.323 -0700 | 2023-05-01 11:42:20.326 -0700 | 2023-08-27 17:35:52.761 -0700 | ACCOUNTADMIN | NULL | +-----------------------+--------+-----------+-----------+-----------------+--------------+----------+-------------------+-------------+--------------+------------+-------------------------------+-------------------------------+-------------------------------+--------------+---------+ Serviceの作成 フロント側、サーバ側の2つのServiceを作成していきます。 specについて、ステージにファイルを配置してそれを指定するスタイルのほかに、 以下のようにCREATE SERVICEに含めるスタイルがあるようです。 CREATE SERVICE フロント側のServiceは以下です。 CREATE SERVICE ikuty_api_service IN COMPUTE POOL ikuty_scs_pool FROM SPECIFICATION $$ spec: containers: - name: api-container image: endpoints: - name: api port: 8080 $$ ; サーバ側のServiceは以下です。 CREATE SERVICE ikuty_front_service IN COMPUTE POOL ikuty_scs_pool FROM SPECIFICATION $$ spec: containers: - name: front-container image: endpoints: - name: front port: 5173 public: true $$ ; SERVICE用のシステム関数 SaaSで動くコンテナの動作を確認するのは結構面倒なことなのかなと思います。 自分の操作に対してSaaS側で何が行われているのか知りたいことは結構あるのかなと思います。 SPCSには以下のコマンドがあるようです。 SYSTEM$GET_SERVICE_STATUS SYSTEM$GET_SERVICE_LOGS エンドポイントURLの取得 SHOW ENDPOINTS すると、Specで指定したpublicなendpointを得られました。 ingress_url に なんとかかんとか.snowflakecomputing.app というURLが入っています。 SHOW ENDPOINTS 動作確認 Computing poolのStatusがACTIVEになってから、エンドポイントURLをブラウザで開くと、 期待通り、Reactで作ったHello Worldアプリが表示されます。 SYSEM$GET_SERVICE_LOGS()でフロントサービスのログを覗くと、viteの起動ログが出ていました。 そうえいば 5173 を 80 に mapping する記述をどこにもしていないのですが、そうなっています。 > front@0.0.0 dev > vite VITE v4.5.2 ready in 314 ms ➜ Local: http://localhost:5173/ ➜ Network: http://10.244.1.3:5173/ ➜ Network: http://172.16.0.6:5173/ 同様に、サーバ側のログを覗くと、uvicornの起動ログが出ていました。 ViteのProxyから8080で繋がるので、こちらは8080が開いています。 INFO: Will watch for changes in these directories: [\'/app\'] INFO: Uvicorn running on http://0.0.0.0:8080 (Press CTRL+C to quit) INFO: Started reloader process [1] using StatReload INFO: Started server process [8] INFO: Waiting for application startup. INFO: Application startup complete. まとめ SnowflakeをWebサーバのインフラにするだけの内容で正直意味がないです。 しかし、APIでSnowflakeに触ったり、Reactで格好良い可視化をしたり、夢は広がります。 FastAPI,React,TypeScriptの恩恵ゼロなので、今後ちょっと凝ったものを作ってみます。

FitbitはAPIがしっかり整備されていて、OAuth2 endpoint経由でデータが取り放題。 せっかくなので、話題のExternal Network Access(2023年12月現在 PuPr)を試してみようと思う。 つまり、FitbitAPI→Snowflakeをやってみようと思う。 Fitbit APIを使用するにはOAuth2.0 Authorizationを通す必要がある。 Snowflakeの公式にOAuth2.0 Endpoint経由でGoogle翻訳APIと連携する段取りが書かれていて、 それをそのままFitbit APIのものに差し替えるだけで動いた。 外部ネットワークアクセスの例 外部ネットワークアクセスについては以下。 [clink implicit=\"false\" url=\"https://docs.snowflake.com/ja/developer-guide/external-network-access/creating-using-external-network-access#label-creating-using-external-access-integration-network-rule\" imgurl=\"https://www.snowflake.com/wp-content/uploads/2017/01/snowflake-logo-color-300x69.png\" title=\"外部アクセス統合の作成と使用\" excerpt=\"特定の外部ネットワークロケーションへのアクセスを有効にするには、外部ロケーションのリストと使用を許可されるシークレットのリストを指定する外部アクセス統合を作成します。UDF の作成時、あるいは CREATEFUNCTION または CREATEPROCEDURE でプロシージャを作成する際に、 EXTERNAL_ACCESS_INTEGRATIONS 句を使用してこの統合を参照することで、ハンドラーコードが外部ロケーションとの認証コードにシークレットを使用できるようになります。\"] 2016年6月に書いた記事。phpで検証をしていた。 この辺りからバッテリーがダメになる度に新しいFitbit Charge(1,2,3)を買って溜めてきた。 この間、FitbitがGoogleに買われてしまったり、スマホアプリが大幅に変わったり、色々あった。 基本的な機能はずっと動いているので、7年分のデータが溜まっているんじゃないかな、と期待。 [clink url=\"https://ikuty.com/2016/06/07/fitbitapi-authenticate-grant-flow/\"] Fitbit API側の準備 OAuth2連携に必要な情報を dev.fitbit.com から取得する必要がある。 Authorization Code Grant Flow with PKCE こちらを参考にさせていただいた。 [clink implicit=\"false\" url=\"https://www.zenryoku-kun.com/post/fitbit-api#register-app\" imgurl=\"https://www.zenryoku-kun.com/home/sakura-400w.jpg\" title=\"FitbitのWeb APIを実行する方法\" excerpt=\"Fitbit Sense2を購入しました。はじめてのスマートウォッチです。Fitbitデバイスでは、心拍数や歩数等、収集したデータをWeb APIで取得することが可能です。さっそく使って遊んでみようと思ったら、Web APIの認証がなかなか通らない、、、ドキュメントはとても充実しているのですが、OAuth2.0の認証パターンがImplicit Grant Flowの場合、Authorization Code Grant Flowの場合、PKCEを使う場合、、、などなど、情報量がとにかく多く混乱してしまいました。何はともあれ、何とか認証を通して、こんな感じで歩数などのアクティビティ情報や、心拍数や血中酸素濃度（SpO2）を取得することが出来ました。\"] 以下を準備すればOK。 access-token refresh-token client-id Snowflakeでリソース作り Snowsightでポチポチとリソースを作っていく。 USE ROLE SYSADMIN; -- 外部ロケーションを表すネットワークルールの作成 -- CREATE OR REPLACE NETWORK RULE fitbit_apis_network_rule MODE = EGRESS TYPE = HOST_PORT VALUE_LIST = (\'api.fitbit.com\'); -- 外部ロケーションとの認証に必要なOAuth認証情報を保持するセキュリティ統合の作成 -- CREATE OR REPLACE SECURITY INTEGRATION fitbit_api_oauth TYPE = API_AUTHENTICATION AUTH_TYPE = OAUTH2 OAUTH_CLIENT_ID = \'\' OAUTH_CLIENT_SECRET = \'\' OAUTH_TOKEN_ENDPOINT = \'https://api.fitbit.com/oauth2/token\' OAUTH_AUTHORIZATION_ENDPOINT = \'https://www.fitbit.com/oauth2/authorize\' ENABLED = TRUE; -- セキュリティ統合に含まれる認証情報を表すシークレットの作成 -- CREATE OR REPLACE SECRET fitbit_api_oauth_token TYPE = oauth2 API_AUTHENTICATION = fitbit_api_oauth OAUTH_REFRESH_TOKEN = \'\'; 最後に外部アクセス統合を作成する。 ストレージ統合や、Notification統合など、統合の作成にはACCOUNTADMINが必要で、 同様に外部アクセス統合の作成にはACCOUNTADMINが必要とのこと。 USE ROLE ACCOUNTADMIN; CREATE OR REPLACE EXTERNAL ACCESS INTEGRATION fitbit_apis_access_integration ALLOWED_NETWORK_RULES = (fitbit_apis_network_rule) ALLOWED_AUTHENTICATION_SECRETS = (fitbit_api_oauth_token) ENABLED = TRUE; 外部ロケーション(ネットワーク)にアクセスするUDFsを書くロールを作成する。 UDFsを書く際に、シークレットを参照する必要がある。 UDFsを書けるロールにシークレットのREAD権限を付与しておく必要がある。 以下、そのままでは SECURITYADMINがDB・スキーマに触れないので環境により修正が必要。 USE ROLE USERADMIN; CREATE OR REPLACE ROLE ikuty_fitbitapi_developer; USE ROLE SECURITYADMIN; USE SCHEMA IKUTY_DB.PUBLIC; GRANT READ ON SECRET IKUTY_DB.PUBLIC.fitbit_api_oauth_token TO ROLE ikuty_fitbitapi_developer; GRANT USAGE ON INTEGRATION fitbit_apis_access_integration TO ROLE ikuty_fitbitapi_developer; GRANT ROLE ikuty_fitbitapi_developer TO role SYSADMIN; 本体の実装 PythonでOAuth2 Endpoint経由でFitbit APIにGETリクエストを投げるFunctionを書く。 最初、トークンのexpire時のrefreshを自力で書いていたが、get_oauth_access_token(\'cred\')により、 自動的にrefreshしてくれていることに気づいた。 use role sysadmin; use schema IKUTY_DB.PUBLIC; CREATE OR REPLACE FUNCTION fitbit_python() RETURNS STRING LANGUAGE PYTHON RUNTIME_VERSION = 3.8 HANDLER = \'hello_fitbit\' EXTERNAL_ACCESS_INTEGRATIONS = (fitbit_apis_access_integration) PACKAGES = (\'snowflake-snowpark-python\',\'requests\') SECRETS = (\'cred\' = fitbit_api_oauth_token ) AS $$ import _snowflake import requests import json def hello_fitbit(): with requests.Session() as s: access_token = _snowflake.get_oauth_access_token(\'cred\') url = \"https://api.fitbit.com/1/user/-/activities/steps/date/today/1m.json\" res = s.get(url,headers={\"Authorization\": \"Bearer \" + access_token}) res_data = res.json() return res_data $$; 実行結果は以下。1日毎の歩数を1ヶ月分取得できた(恥...)。 select parse_json(fitbit_python()); { \"activities-steps\": [ { \"dateTime\": \"2023-11-23\", \"value\": \"15570\" }, { \"dateTime\": \"2023-11-24\", \"value\": \"5392\" }, { \"dateTime\": \"2023-11-25\", \"value\": \"8993\" }, { \"dateTime\": \"2023-11-26\", \"value\": \"10525\" }, { \"dateTime\": \"2023-11-27\", \"value\": \"6371\" }, { \"dateTime\": \"2023-11-28\", \"value\": \"2713\" }, { \"dateTime\": \"2023-11-29\", \"value\": \"9252\" }, { \"dateTime\": \"2023-11-30\", \"value\": \"0\" }, { \"dateTime\": \"2023-12-01\", \"value\": \"7947\" }, { \"dateTime\": \"2023-12-02\", \"value\": \"11265\" }, { \"dateTime\": \"2023-12-03\", \"value\": \"8557\" }, { \"dateTime\": \"2023-12-04\", \"value\": \"2366\" }, { \"dateTime\": \"2023-12-05\", \"value\": \"7985\" }, { \"dateTime\": \"2023-12-06\", \"value\": \"8109\" }, { \"dateTime\": \"2023-12-07\", \"value\": \"6852\" }, { \"dateTime\": \"2023-12-08\", \"value\": \"3707\" }, { \"dateTime\": \"2023-12-09\", \"value\": \"12640\" }, { \"dateTime\": \"2023-12-10\", \"value\": \"7122\" }, { \"dateTime\": \"2023-12-11\", \"value\": \"7190\" }, { \"dateTime\": \"2023-12-12\", \"value\": \"8034\" }, { \"dateTime\": \"2023-12-13\", \"value\": \"5228\" }, { \"dateTime\": \"2023-12-14\", \"value\": \"2861\" }, { \"dateTime\": \"2023-12-15\", \"value\": \"6785\" }, { \"dateTime\": \"2023-12-16\", \"value\": \"11720\" }, { \"dateTime\": \"2023-12-17\", \"value\": \"11021\" }, { \"dateTime\": \"2023-12-18\", \"value\": \"0\" }, { \"dateTime\": \"2023-12-19\", \"value\": \"11021\" }, { \"dateTime\": \"2023-12-20\", \"value\": \"0\" }, { \"dateTime\": \"2023-12-21\", \"value\": \"2703\" }, { \"dateTime\": \"2023-12-22\", \"value\": \"3336\" }, { \"dateTime\": \"2023-12-23\", \"value\": \"7497\" } ] } 結論 PuPrのExternal Network Accessを使用して、FitbitAPI→Snowflakeが出来ることを確認した。 (途中、自動的にトークンをrefreshしてくれている、と書いたが、何度かExpireさせないと良くわからない。) 相手がOAuth2.0ならとても簡単に繋ぐことができると思う。 次は、せっかくなのでSiS(Streamlit in Snowflake)で可視化してみたりしたい。

この記事はSnowflake Advent Calendar 2023シリーズ2の19日目です。 今年はSnowProAdvanced: Architect試験に合格できました。 結局のところ資格試験であるという側面はあるものの、いろいろ役立っている実感があります。 その後、Mediumというメディアで気になる記事を読み漁る、みたいなことを始めました。 正直知らないことばかりです..。 いくつか読んだ記事のうち、これはヤバいなと感じた記事の読書感想文を書こうと思います。 [clink implicit=\"false\" url=\"https://medium.com/snowflake/deep-dive-into-the-internals-of-snowflake-virtual-warehouses-d6d9676127d2\" imgurl=\"https://miro.medium.com/v2/resize:fit:1002/format:webp/0*6KqDj8Y_HxeL11xT.png\" title=\"Deep dive into the internals of Snowflake Virtual Warehouses\" excerpt=\"Snowflake’s Data Cloud provided as Software-as-a-Service (SaaS), enables data storage, processing, analytic solutions, Machine Learning, and running apps & services in a performant, easy-to-use, and flexible manner using “virtual warehouses” which is the primary compute primitive in Snowflake. This post details the internals of virtual warehouses which provide elastic, highly available, and fully managed mechanisms to run a variety of customer workloads on multiple cloud service providers.\"] 訳は間違っているところもあると思います。ご容赦ください。 [arst_toc tag=\"h4\"] 仮想ウェアハウスの基本 まず、コンピュートとストレージが分離し、それぞれ独立してスケールできることが特徴としている。 Snowflakeにおいて、仮想ウェアハウスはコンピュートの最小単位ではあるが、仮想ウェアハウスは 複数のVMからなるMPPクラスタであると言及している。 この記事は、仮想ウェアハウスを説明するために仮想ウェアハウスを構成するVMに言及している。 仮想ウェアハウスの下に物理のVMがいることにフォーカスがあてられている。 SnowflakeのSaaSサービスを実現するコードはMPPクラスタを構成する各VMで動いていて、 ジョブ実行の際、各VMはリソースを直接参照するしVM同士でmeshN/Wを構成して資源を共有する。 (後述) 仮想ウェアハウス同士はストレージを共有しないけれど、仮想ウェアハウス内部のVMは むちゃくちゃ密に連携しあって、計算資源もストレージも共有しあう。 このセクションで、仮想ウェアハウスの設計方針が述べられている。 「可能な限り顧客に選択肢を提供するのを避けSnowflakeがベストを考える」が基本方針である一方、 「仮想ウェアハウスを構成するVMの物理資源を変更できる柔軟性を提供する」と言っている。 以降、仮想ウェアハウスを構成するVMの振る舞いについて書かれている 仮想ウェアハウスのサイズとタイプ 仮想ウェアハウスのタイプはCPUとメモリの比率、サイズはCPUとメモリの総量を決める。 タイプは、StandardとSnorpark-optimizedの2種類。 Snowpark-optimizedは、Standardの16倍のメモリ量と10倍のSSDを持つ。 メモリ増量により計算が高速化する。ストレージが大きいとキャッシュや中間生成物が 後続の実行で再利用され高速化する。 中間生成物の書き込みに対し、第1に仮想ウェアハウス上のVMのメモリが使われる。 メモリを使い切ったとき、VMのローカルSSDが使われる。 SSDも使い切ったとき、S3等のリモートストレージが使われる。 QUERY_HISTORY viewにSSD、リモートストレージにスピルした量を出力するので、 メモリが溢れないようにするか、少なくともSSDには乗るようにサイズを増やせよ、と言っている。 (やはりストーリーがストレートでわかりやすい..) SELECT QUERY_ID ,USER_NAME ,WAREHOUSE_NAME ,WAREHOUSE_SIZE ,BYTES_SCANNED ,BYTES_SPILLED_TO_REMOTE_STORAGE ,BYTES_SPILLED_TO_REMOTE_STORAGE / BYTES_SCANNED AS SPILLING_READ_RATIO FROM \"SNOWFLAKE\".\"ACCOUNT_USAGE\".\"QUERY_HISTORY\" WHERE BYTES_SPILLED_TO_REMOTE_STORAGE > BYTES_SCANNED * 5 - Each byte read was spilled 5x on average ORDER BY SPILLING_READ_RATIO DESC ; マルチクラスタ ウェアハウス マルチクラスタは、ジョブの同時実行性を高めるためにクラスタを静的/動的に追加する仕組み。 クラスタ内のVMは相互に関係し合いリソース共有して複数台でジョブのオフロードを行うため、 単一クエリのパフォーマンスアップに寄与する。一方で、クラスタ間はリソース共有しないため、 増えたクラスタ内のVMはジョブのオフロード先の融通にはならず、同時実行時の性能劣化予防に働く。 他にスケーリングポリシーの話や、Min/Max設定による静的/動的追加の話が書かれているが省略。 UpではなくOutの方が費用対効果が高い例として、interleaved workloadsが挙げられている。 Outで増やしたクラスタがダラダラと回り続けるケースが除外できず理論値ではあるけれども、 Upに対するOutのメリットを言う場合に説明しやすい図だなと思った。 この辺りモヤモヤしていたのでバシっと説明してもらえて助かりました。 柔軟性-ステートレスなスケーリング 需給調整の文脈ではなく、自動起動と自動サスペンドの文脈で仮想ウェアハウスの状態が書かれている。 リソースがステートレスであれば、需要の増減と関係なくリソースを増減できる。 仮想ウェアハウスはステートレスリソースであって、需要の発生によりプロビジョンングされ、 需要の消滅により仮想ウェアハウスに紐づくリソースが破棄される。 仮想ウェアハウスにジョブが送信されると、クラスタ内のVMはジョブ実行中にのみ存続するプロセスを 生成する。プロセスが失敗した場合、自動的に再試行される。 ユーザとウェアハウスは多対多の関係であり、ウェアハウスから見ると同時に複数の需要が発生する。 異なる組織・部署がウェアハウスを使用するケースにおいて、ウェアハウスは同時にそれぞれを処理する。 各々のウェアハウスは同じ共有テーブルにアクセスできるが、その際、データのコピーをウェアハウス内に 持たなくても良いように作られているので、各組織・部署の処理が他の組織・部署に暴露されるリスクを 回避できるようになっている。 異なる組織・部署が実行したジョブがウェアハウス上で相互作用しない、という事実があり、 組織・部署から見れば、他の組織・部署に全く影響されず自由にウェアハウスを利用できるという 書き方になっていて、ちょっと抽象度が高いですが「ステートレス」が説明されていました。 柔軟性-マルチクラスタ オートスケーリング スケーリングポリシーの説明。 スケーリングポリシーの設定により、各クラスタの自動起動・シャットダウンの相対的な速度を制御する。 スタンダードポリシーはクレジット消費削減よりもクラスタ追加を優先し、クエリ所要時間を最小化する。 エコノミーポリシーの設定により、クラスタを追加するよりも現在実行中のクラスタを全開で回すことが 優先され、結果としてクエリがキューに入りやすくなり所要時間が延びるが、クレジット消費は減る。 この説明は公式通り。 柔軟性-ゼロへのスケール Auto-resumeとAuto-suspendの説明。 ウェアハウスに対する需要がなくなって一定期間経ったら自動的に停止する。 ウェアハウスに対する需要が発生したら自動的に再開する。その時間等を調整できる。 これらの設定はクラスタではなくウェアハウスに対して設定する。 これも説明は公式通り。需要がなくなったら1個も起動していない状態にできることが主張ポイント 柔軟性-自動Suspend期間の管理 Suspendは、つまり仮想ウェアハウスを構成するVMのリリースなので、VMが持つSSDに 蓄えられたキャッシュは同時に破棄されてしまう。これは、後続のジョブが発生したときに クエリ結果キャッシュが効かなくなることに繋がる。 公式の通り、「ウェアハウス稼働時間(クレジット消費)」と「クエリパフォーマンス」がトレードオフの 関係となる。需要がなくなってすぐにウェアハウスを止めると確かにクレジット消費は減るが、 キャッシュヒット率が下がる。トレードオフにSweet spotがあるので探しましょうと書かれている。 これに留まらず、どういう風に決めたら良いかガイドが書かれている。 ただ、これは答えが無い問題で、実験してねとも書いてある。 - タスク実行、ロード、ETL/ELTユースケースにおいて、すぐに止めた方が良い。 - BI等SELECTが起きるユースケースは、止めるまで10分待つべき。 - DevOps,DataOps,Data Scienceのユースケースは、停止時間は5分が最適。 とりあえず、タスク実行、ロードでは、自動Suspend期間を持たせる意味はないので、 そこは、バッサリ最速で落とす勇気が出る書き方で参考になりました。 全てのクエリのうち、SSDからスキャンした割合を集計するクエリは以下。 この割合が低いということは、ウェアハウスのSuspendが早すぎることを示している。 SELECT WAREHOUSE_NAME ,COUNT(*) AS QUERY_COUNT ,SUM(BYTES_SCANNED) AS BYTES_SCANNED ,SUM(BYTES_SCANNED*PERCENTAGE_SCANNED_FROM_CACHE) AS BYTES_SCANNED_FROM_CACHE ,SUM(BYTES_SCANNED*PERCENTAGE_SCANNED_FROM_CACHE) / SUM(BYTES_SCANNED) AS PERCENT_SCANNED_FROM_CACHE FROM \"SNOWFLAKE\".\"ACCOUNT_USAGE\".\"QUERY_HISTORY\" WHERE START_TIME >= dateadd(month,-1,current_timestamp()) AND BYTES_SCANNED > 0 GROUP BY 1 ORDER BY 5 ; 柔軟性-ウェアハウス内のVMは起動済みVMのプールから割り当てられる VMをコールドから起動するには10秒オーダーの時間がかかる。そもそも小規模のクラウドサービスでは VMの数が不足して流動性がない場合もあり、起動済みのVMをプールして再利用することで、 これらの問題を解決しようとしている。 Snowflakeは、VMの起動、終了、停止、再開、スケーリング等のオペレーション時間に対して、 内部でサービスレベル目標を設けている。 (これらの時間がサービスレベル目標から外れるとSnowflake内部でインシデント管理されるらしい。) ユーザのリクエストで需要が発生した場合、起動済みVMのプールからVMが選ばれ、 ウェアハウスに割り当てられる。 起動済みVMのプールのサイズは、過去の需要のベースラインとスパイクから予測されているらしい。 確かにウェアハウスが瞬時に起動する仕組みが気にはなっていました。 妥当な仕組みで成立しているようですが、言及されている点がポイントかと思います。 柔軟性-需要のバーストに対して用意されるQAS サイズアップの他にQAS(Query Acceleration Service)というサービスが存在する。 起動済みVMプールにあるVMを需給に応じて自動的にウェアハウスに組み入れる。 ウェアハウス内でVMは密に連携してクエリをオフロードし合う。 動的なサイズアップであって、疎連携のマルチクラスタとは異なる。 QASは主に、巨大なテーブルのScanや、burstyなワークロードを目的とする。 QASを使用すると、大規模なクエリが検知された場合にウェアハウス内のVMが ウェアハウスから離れ、他のユーザの小規模なクエリに使われるらしい。 通常はウェアハウスのサイズアップよりも低いコストで目的を達成できるそう。 この手の機能が何故ワークロードを高速化するのか、結局のところ中身を知らないとわからないと 思うので、機能の説明の他に、どういう作りなのかを書いてくれるととても参考になる気がする。 When to useはburstyなワークロードということ。 QASで恩恵を受けられるクエリがどれぐらいあるか気になるところ。 公式によると以下の特徴を持つクエリはQASの恩恵を得られないそう。 フィルターや集計（つまり、 GROUP BY）がない。Query Acceleration Serviceは現在、このようなクエリを高速化できません。 フィルターの選択性が十分ではない。または、 GROUP BY 式のカーディナリティが高くなっている。 十分なパーティションがない。スキャンするために十分なパーティションがないと、クエリアクセラレーションの利点は、サービス用に追加のサーバーを取得する際の待機時間によって相殺されます。 クエリに LIMIT 句が含まれている。ただし、 ORDER BY 句を含んでいる LIMIT 句はサポート されます。 QASの恩恵を得られるクエリとウェアハウスは以下のビューから探すことができる。 -- アクセラレーションの対象となるクエリ実行時間の量によって、 -- サービスから最もメリットを受ける可能性のあるクエリを識別します。 SELECT query_id, eligible_query_acceleration_time FROM SNOWFLAKE.ACCOUNT_USAGE.QUERY_ACCELERATION_ELIGIBLE ORDER BY eligible_query_acceleration_time DESC; -- Query Acceleration Serviceの特定の期間中、 -- 対象となるクエリが最も多いウェアハウスを識別します。 SELECT query_id, eligible_query_acceleration_time FROM SNOWFLAKE.ACCOUNT_USAGE.QUERY_ACCELERATION_ELIGIBLE WHERE warehouse_name = \'mywh\' ORDER BY eligible_query_acceleration_time DESC; QASにより、ウェアハウスは需給調整のためにVMをリース(借りる)する、という表現がある。 ウェアハウスがリースできるVMの数の最大値は、Scale Factorという数値で表される。 要は、通常のウェアハウスサイズで確保するVMの数の何倍のVMをリースできるか。 例えば、Scale Factorが5、VMのサイズがM(つまり4credsits/hour)の場合、 4*5=20 credits/hourまで増強することになる。 Scale FactorはQUERY_ACCELERATION_ELIGIBLEビューにあり、 クエリID単位で知ることができる。 SELECT MAX(upper_limit_scale_factor) FROM SNOWFLAKE.ACCOUNT_USAGE.QUERY_ACCELERATION_ELIGIBLE WHERE warehouse_name = \'mywh\'; 仮想ウェアハウスのジョブスケジューリング スループット最大化、レイテンシ最小化、クラスタ使用率最大化、異なる需要に対して供給のために、 ウェアハウスの負荷を追跡・調整するウェアハウススケジューリングサービス(WSS)が備わっていて、 クエリがクラウドサービスレイヤでコンパイルされた後、WSSがジョブスケジューリングを行う。 WSSは各VMのCPU・メモリ使用量を追跡する。ウェアハウスのメモリキャパシティは、 各VMの実効メモリ(OSやソフトウエアの使用を除く)にウェアハウス内のVMの数を掛けたもの。 メモリが使い果たされたことを検知して、データをdiskに吐き出す(Spill)。 メモリ負荷が高くなりすぎると、VMは落とされて\"リタイア\"(前述)する場合がある。 情報科学の用語の1つにDOP(Degree Of Parallelism)がある。 WSSは1個のジョブを何個のプロセスで同時処理して完了するか、という制御を行なっているらしい。 VMのCPUコアが1つのプロセスを受け持ち、CPUコアの数だけプロセスを並列実行できる。 例えばCPUコアを8個もつVMを4個もつウェアハウスの保持コア数は合計32個。 1つのジョブを32コアで並列処理しても良いし、逆に32個のジョブを1コアで処理しても良い。 DOPはコンパイル時に推定される。 以降、ジョブスケジューリングの少し詳しい説明が書かれている。 実行中の各ウェアハウスは既にキューにジョブが積まれている。 その上で新しいジョブを処理する場合、どのウェアハウスで処理すべきかを決めることになる。 WSSはウェアハウスの全てのVMに均等に負荷分散されるべき、という仮定を立てる。 クラウドサービスレイヤは、ジョブの処理に必要なメモリとコンパイル時に決まったDOPから、 そのジョブをどのウェアハウスで処理するかを決める。 メモリの使用状況や同時実行性(?、キューに積む時点でジョブがどれぐらい並列実行されているか??) を見て、ウェアハウスの適格性を決める。適格性が同じなら、その時点で同時実行ジョブが最も少ない ウェアハウスを選択する。適格なウェアハウスが無い場合、WSSキューに残り続ける。 ジョブスケジュールを行うと、各ウェアハウスのリソース使用状況バランスが変化する。 WSSはクラウドサービスにVM使用状況のレポートを送る。 クラウドサービスは状況次第でDOPを下げる(より少ない並列度で処理するよう計画される)。 DOPを下げた後、ジョブはウェアハウスで実行される。ジョブ終了後リソースは解放される。 負荷に応じてDOPがダイナミックに調整されている様が書かれている。 実際のところ、DOPの推移を観察することはできないのと、DOPの上げ下げとパフォーマンスの 関連が本当にその通りなのか不明なこともあり、結局良くわからない。 並列レベルの制御 MAX_CONCURRENCY_LEVELパラメタにより、最大並列処理数を設定できる。 デフォルト値は8ということなので、最大で4個のジョブを並列実行することになる。 巨大なクエリを処理する場合、1個のジョブを受け持つコア数を増やすことでスループットが上がる 場合があるらしい。並列処理数が下がるとキューに積まれるジョブが増えることに繋がる。 ウェアハウスサイズを増やさずにMAX_CONCURRENCY_LEVELだけ調整しても、 リソースの総量は変わらないはずだし、簡単に最適値が見つかるなら全自動で決めてくれる のだろうから、きっと難しい話なのだろう。QASみたいに全然違う何かを使うと良いよ、と書かれている これは公式の以下のドキュメントが対応する。 同時実行クエリの制限 リソースモニタと使用量制限 クレジットを想定よりも多く消費しないようにするアラートとハードリミットの仕組み。 消費クレジットが制限を超えたことをトリガにアラート、自動停止を実行できる。 リソースモニタが設定されていないウェアハウスを以下のクエリで見つけて設定せよとのこと。 SHOW WAREHOUSES ; SELECT \"name\" AS WAREHOUSE_NAME ,\"size\" AS WAREHOUSE_SIZE FROM TABLE(RESULT_SCAN(LAST_QUERY_ID())) WHERE \"resource_monitor\" = \'null\' ; ウェアハウスの負荷とサイズの決定方法 Snowsightでウェアハウスの負荷を確認できる。これの計算方法などが書かれている。 確かに、あれ、何をどうやって集計したチャートなのか知らなかった。 Snowflakeが出力するメトリクスを見てウェアハウスの正しいサイズを決定せよとのこと。 ウェアハウスのジョブ負荷メトリクスは、一定期間内の実行ジョブ数、キューに入ったジョブ数の 平均である、とのこと。実行ジョブ数の平均は、全てのジョブの実行時間(秒)を期間(秒)で 割った値であるとのこと。これはバーの青色の部分だな。 Private Previewで、ウェアハウスの使用率メトリクスが用意されるらしい。 以下の表のように、ウェアハウス単位、クラスタ単位で100分率の値を得られる。 ウェアハウス負荷や使用率によって、キャパシティ割り当てを行うべきとのこと。 どういう数字だったらどうすべきか書かれている。そういえば知らなかった。 ワークロードのスループット・レイテンシが適切で、キューに入ったクエリが少なく、 長期にわたりクエリ負荷が1未満、かつ、使用率が50%を切る場合、 ウェアハウス・クラスタのダウンサイズを検討する。別のウェアハウスを起動し、 キューに入れられたジョブをそのウェアハウスで実行できるようにする。 ワークロードのスループット・レイテンシが期待よりも低速で、かつ、 クエリ負荷が低く、かつ、使用率が75%を超えるなど高い場合、 ウェアハウスのアップサイズを検討するか、クラスタの追加を検討する。 使用量の急増(スパイク)が繰り返し発生する場合、 ウェアハウスの追加・クラスタの増量を行い、スパイクに対応するクエリをそれに移す。 スパイク以外のクエリを小さいウェアハウス・クラスタで実行されるようにする。 ワークロードが通常よりも大幅に高い場合、 どのジョブが負荷に寄与しているのか調査する。 ウェアハウスが定期的に実行される(スパイクではない)が、かなりの期間にわたって 合計ジョブ負荷が1未満である場合、 ウェアハウスのサイズダウン、クラスタの削減を検討する。 ストレージ・キャッシュ-ストレージアーキテクチャ Snowflakeには、テーブルの永続化、JOIN等のクエリ演算子によって生成されクエリの実行中に消費される 中間データの2つの形式のストレージがある。 永続化テーブル 寿命が長い永続化テーブルは、S3等のオブジェクトストレージが使われる。 オブジェクトストレージは比較的スループットが高くないが、長期間保管する際の可用性要件が良い。 S3等のブロックストレージに対して一括上書きすることになるが、immutableなデータを 扱うには適している。ブロックストレージの上でimmutableなデータの水平展開を行う。 (別のMedium記事で、micro-partitionはテーブルのバージョニングであって、immutableな データ領域を重ねていくことと、その仕組みにより副作用的にTime-Travelが用意されることが 書かれている。micro-partitionがブロックストレージ上で増えていく様は面白い) immutableなファイルには列データ、属性データがグルーピング・圧縮され格納されている。 相対位置が付与されていて再構成しやすい。 ブロックストレージに備わっている「部分的な読み取り」機能により、これらのファイルの 必要な部分を取得する。こうして永続化テーブルがブロックストレージに保管・使用される。 JOIN等のクエリ演算子によって生成されクエリの実行中に消費される中間データ 中間データは寿命が短く低レイテンシ・高スループットが求められる。 ジョブの実行にウェアハウスのメインメモリとSSDが使われる。 これらはウェアハウスの開始時に作られ、終了時に破棄される。 これらの一時ストレージは、リモートにある永続化テーブルのライトスルーキャッシュとして機能する。 各仮想ウェアハウスはそれぞれ個別に一時ストレージを持ち、クエリ実行時に使用される。 この一時テーブルは、全ての仮想ウェアハウスから\"個別にコピーすること無しに\"共有できる。 メモリ管理を単純化するためのSpill 中間データの書き込み操作の際に、まずウェアハウス内のメインメモリが使われる。 メインメモリがfullになると、ウェアハウスのローカルdisk(SSD)が使われる。 ローカルdiskがfullになると、リモートストレージが使われる。 メモリ不足、ディスク不足を回避するための仕組みになっている。 事実としては良く知られた挙動だけれども、それと「メモリ管理の単純化」というストーリーが 紐づいて理解しやすくなった気がする。 ストレージ・キャッシュ-キャッシュ戦略 「キャッシュ」とは、良く使うデータを取り出しやすいところに一時的に保存しておくもの。 キャッシュ容量は限られるため、ヒット率を維持しつつ効率的に中身を更新することが重要。 その具体的な仕組みとして、LRU (Least Recently Used)、LFU (Least Frequently Used)が有名。 キャッシュが必要な中間データ(前述)量が小さい場合、一時ストレージレイヤ(=VMのdisk)は、 ファイル名のハッシュ値を使ったLRUキャッシュにより、頻繁にアクセスする永続化データの キャッシュとして使われる。このキャッシュは低優先度で\"lazy\"に行われるらしい。 ファイルが仮想ウェアハウスのどのVMにストアされるかについて「一貫性」が言われている。 一方向関数にファイル名を食わせた結果、ファイル名とストア先VMが決まることを言っている。 サイズ変更によってVMの追加・削除が行われる際にキャッシュがシャッフルされてしまわない。 (VMのサイズが同じならば)永続化ストレージ上のファイルは特定のVMに保存されるため、 永続化ストレージ上のファイルに対する操作は、そのファイルのハッシュが保存されるVMが 実行するようにスケジューリングされる。こうして、ジョブの並列化はファイルのハッシュ値が 一貫して同じVMに保存されることと密接に結びついている。 ファイル名が偏っているとハッシュも偏り、保存先のVMが偏る場合がある。 それを回避するため、ワークロードがそのVMでの所要時間が他のVMでの所要時間よりも 小さいかどうか、に基づいてクラスタ内のVM内でロードバランシングが行われる。(え..?) キャッシュ(execution artifacts)が移動した場合(キャッシュアウトした場合)、 最初に実行がスケジュールされていた既に過負荷になっているVMの負荷がさらに増加する のを避けるため、操作の実行に必要なファイルが永続化ストレージから読み取られる。 仮想化の問題、ネットワークの問題など様々な理由で一部のVMが極端に遅い時があるらしい。 その対策にもなっているらしい。 Snowflakeのスケジューリングロジックは、execution artifactsを永続化ストレージ のキャッシュ先と同じVMに配置することと、全てのexecution artifactsを少数のVMに 配置することの間のバランスを見つけようとする。 前者は永続化ストレージのReadに伴うネットワークトラフィックの最小化を目指すが、 ジョブがウェアハウス内の全てのVMにスケジューリングされることによって中間データが VM間でやり取りされることに起因してネットワークトラフィックが増加するリスクもある。 後者は中間データ交換のためのネットワークトラフィックがなくなる(減る..?)が、 永続化ストレージのReadのためのネットワークトラフィックが増加する可能性がある。 一時データ容量はリモートの永続化ストレージ容量よりもかなり小さい(平均0.1%未満) にも関わらず、Snowflakeのキャッシュスキーム上では、Readのみのクエリで-80%、 Read-Writeがあるクエリで-60%のキャッシュヒット率にもなるらしい。 文章だけでは読みきれないな..。ただキャッシュの仕組みが書かれているだけでなく、 永続化ストレージ上のデータ(=ファイル)をVMに持ってくる仕組みの説明になっていて、 ウェアハウス内のVMで負荷分散して処理していく様が薄ら分かった気がする。 マルチテナント環境におけるセキュリティとリソース分離 アカウント、ジョブごとにデータを分離し、アカウント、ジョブ間でデータが漏洩しないように 設計している。\"仮想マシンを分離すること\"により、各テナントの分離を実現している。 さらに、cgroup、カーネル名前空間、seccomp(※)のようなDockerコンテナに似たカーネルプリミティブ を備えたVM内のサンドボックスにより、同一顧客アカウント内のジョブ間の情報漏洩を防ぐ。 ※cgroup,カーネル名前空間,secompはLinuxカーネルの機能で、 Dockerコンテナの内部で使われている。 cgroup,namespaceは、プロセスグループのリソース(CPU、メモリ、ディスクI/Oなど)の利用を 制限・隔離するLinuxカーネルの機能とのこと。seccompは自プロセスが発行するシステムコールを 制限してプロセスを乗っ取られたとしても被害を最小限にする機能とのこと。 各VMを独自のハードウェア、ページテーブル、カーネルを使用して動作させることで、 マルチテナントセキュリティとリソース分離を図っている。 VMが同じハードウェア、ページテーブル、カーネルを使用した\"VM分離\"がない場合、 従来から使われているカーネルカーネル共有方式(cgroup,名前空間,secomp付き)だけでは、 Snowflakeのセキュリティ基準に達しないと判断したそう。(そうですか..)。 \"VM分離\"するよりもカーネルを共有した方が、コンテナは高速に起動して都合が良いけれども、 カーネルを共有するということは、過去のCVEsから予想されるセキュリティ脆弱性に曝露される ことになる。 仮想ウェアハウスを構成するVMはそのウェアハウスが占有するプライベートなリソースであって、 仮想ウェアハウス間で共有されたりはしない。加えて仮想ウェアハウスはステートレス。 データの状態に影響されず、需要に応じてどんな時でも作成・破棄・リサイズできる。 その仕組みのため、ジョブが特定の仮想ウェアハウスで限定して実行されるから、 その仮想ウェアハウスのパフォーマンスが他の仮想ウェアハウスのパフォーマンスに影響しない。 ジョブ実行の際、各仮想ウェアハウス内のVMが新しいプロセスを起動する。 そのプロセスはジョブの実行期間中にのみ生存する。 プロセスの失敗は自動的に検知され即座に修正(再実行)される。 ユーザは、いつでも複数の仮想ウェアハウスを実行できる。 各ウェアハウス上で、複数のジョブが並列実行する。 ネットワークセキュリティ 仮想ウェアハウスは次の外部ネットワークアクセスを必要とする。 クラウドサービスレイヤとの通信 ジョブ実行時に発生する他の仮想ウェアハウスとのデータ共有 ローカルのクラウドストレージ(diskのspill先)へのアクセス API Gatewayへのアクセス Snowflakeは全ての仮想ウェアハウスからのネットワークトラフィックを信用しない。 内部サービスへのトラフィックは必ず認証済みのエンドポイントを経由する。 外部ネットワークへのトラフィックは外向きプロキシを経由し、アクセス制御ポリシーが適用される。 未認証のエンドポイントへのアクセスはブロックされ、予期しない動きはSnowflakeに報告される。 アカウント間で予期しない漏洩が起こらないように、VM、proxy、ジョブ間でやり取りされる全ての 通信が正常であることを、クラウドサービスレイヤがIPアドレスマッチングを行うことで検証する。 仮想ウェアハウスが持つ署名済みの共有シークレットを使って、仮想ウェアハウス間の全ての通信 について、発信・着信側が本当にSnowflake内部の仮想ウェアハウスであるか検証する。 そもそも仮想ウェアハウスからクラウドサービスレイヤへの通信がむちゃくちゃ多くなり、 DoS攻撃のようにならないように、通信にレートリミットがついていたりするらしい。 他には、フローログを使って何かをしているらしい。フローログって何か知らなかったので調べた。 NWインターフェース間で行き来するIPトラフィックに関する情報をキャプチャする機能。とか。 Wireshakみたいなやつだろうか。例えば、仮想ウェアハウス内のVMが知らないdestに対して 送ったIPトラフィックを見つけてforensic inspectionを行いVMを隔離するなど。 ※デジタルフォレンジック。「証拠保全」みたいな使われ方をしている。 うーん..難しい... ネットワークセキュリティと言うと、つい外部から内部(Ingress)の事かなと思っていたが、 SaaSの内部で好き放題されてしまうリスクがある気持ちを理解した。 外部ネットワークアクセスはこの気持ちの上に成立しているんだろう。 Python/Scala/Javaコードの分離 SQLみたいに出来ることが制限されている言語とは違い、何でもできるJava/Python/Scalaで UDFやプロシージャを書くことはセキュリティ面でリスクがいっぱい。 これらの言語で書いた処理は、パフォーマンスの観点で、ジョブの他の処理と同じVM上で動く。 マルチテナント環境上で(処理を?)分離するために(前述のように再利用できない)VMを使用する のに加え、cgroups, namespaces, secomp, eBPF, chrootのようなLinuxカーネル の要素を使ったセキュアなサンドボックスを提供することで、ジョブに割り当たったスコープの外の 情報にアクセスしたり、処理がSnowflakeの他の機能に影響したりしないようにしている。 (これらは前述されている。それぞれうっすら調べてみた。こういう風に作るんだなぁと面白い) Java/Python/Scalaで書かれた各ジョブには、実行用に新たにサンドボックスが割り当てられる。 コードの実行に最低限必要なread-onlyのソフトウエアが用意される。 サンドボックス用のchrootが用意され(/より上に行けない)、その下には書き込み可能ディレクトリが いくつかあるだけ。ジョブはそこで処理を行う。read-onlyなディレクトリがマウントされて、 JavaのJARパッケージ、Pythonパッケージや、データファイルはそこで共有される。 サンドボックス内のジョブ(のリソースを使用するプロセス)はcgroupが設定され、 使用メモリ、CPU使用量、PID使用量(プロセス数?)が制限される。 マルチプロセッサユースケース(マルチスレッド化してプロセス内で処理を並列化する話?)のため スレッド生成がサポートされる。 さらに、許可リスト(IPC,Inter Process Communicationに関するリソースを隔離する仕組み= IPC Namespace、eBPF,extended Berkley Packet Filter=カーネル内で発生した イベントで駆動する処理を安全・簡単に組み込む仕組みによって、予め許可していないartifacts がサンドボックスの外に接続するUNIXソケットを開けないようにする)によるネットワークアクセスの 制限、process namespaceによるVM上の他のプロセスを見えなくする制限、 seccomp(子プロセスのフォーク、実行可能プログラムの実行)によるカーネルAPIの不必要な 実行の回避が行われる。脅威検知のためptraceがシステムコールを管理する。 ジョブが完了した後、VM上の環境のもろもろの解放、開いたソケットのクローズ、 クレデンシャルの削除、ローカルキャッシュ、一時ファイル、ログの削除が行われる。 追加の多層防御手段?(defense-in-depth measure?)として、規定時間内に終了しなかった Python/Job/Scalaコードを実行するプロセスに対して、監視プロセスがkillシグナルを送る。 サンドボックス外に離脱したり、攻撃者が仮想ウェアハウス上のVMにプロセスを残したり ルートキットを配置する未知のリスクに備えて、Python/Java/Scalaコードを実行したVMは 「実行不可」としてマークされる。仮想ウェアハウスのスケジューリングや起動済みVMをプールする 仕組みの上で、Python/Java/Scalaコードを実行したVMが異なるアカウント・ユーザに 割り当てられると、アカウント間情報漏洩のリスクに繋がってしまうため、異なるアカウント・ユーザに 割り当たらないようになっている。Python/Java/Scalaコードを実行するVMが作られると、 アカウント専用のVMプール入れられる。新しいVMを割り当てるときは、まずはアカウント毎の空き プールからVMが選ばれる。 多数のゼロデイエクスプロイト（脆弱性が発見されてからパッチが当たるまでの期間の攻撃)が 連続して使用されると、サンドボックスが破られてしまうかもしれないが、それに備えた作りに なっている。まずエクスプロイトは、ユーザアカウントで実行中のVMに存在する。このVMは、 Snowflakeサービスや、Snowflake内のローカルネットワーク上のVMから隔離されている。 攻撃者が手にしたクレデンシャルは(サンドボックスを破壊した)特定のアカウントの特定のVMに 限定され他では使用できない。 あくまで論理的な構成が書かれているだけで「コンテナ」というワードも無いし、何かチラチラとするな。 こういうのを「コンテナエスケープ」とか言うらしい。 ソフトウエア更新の管理 Snowflakeの各機能がどうやって仮想ウェアハウスにデプロイされるかについて。 (デプロイの)ワークフローにより新機能、セキュリティアップデート、機能改善が行われる。 全ての処理は自動化されていて手作業の間違いが起きないようにしている。 このリリースプロセスにおいて、単体テスト、回帰テスト、結合テスト、性能、負荷テストが行われる。 リリースプロセスは、本番の前段の環境、または本番に近い環境で行われる。 VMがフリープールに入る前に最新のパッチが当たる。VMのStartやResumeなどの操作の後に、 フリープールからVMに割り当たったり、逆にVMからフリープールに抜けたりするが、 フリープールからVMに割り当たるプロセスの一部として、VMに最新に保つための最新のバイナリが ダウンロードされ、適用される。 Resume、Startなどのライフサイクル操作は即座に終わるように作られているが、 影響を与えないように性能要件が与えられているらしい。 SKU sizeやOSのメジャーパーションなど大きな変更の際には、未適用のVMと適用済みのVMの両方が 同時に動く状態となる。古い方は既存のジョブを実行し、新しい方は、新しいジョブを実行する。 そのようにジョブがルーティングされる。 既存のジョブを実行し終わってから、最終的に古い方は消される。 つまり、1個のウェアハウスについて、アップデートの時期を迎えると背後で(適用前後の)2個になる。 前述のようにキャッシュはVMのローカルディスクなので、もし古いウェアハウスが破棄されたとすると、 キャッシュが失われることになる。 それによりキャッシュミスが発生しパフォーマンスに影響しないように、事前に管理されているとのこと。 がんばってテストしているけれども運用環境にバグが混入することもある。 なのでアップデートをロールバックできるようになっている。 クラウドプロバイダのリージョン毎に、動作中のバイナリの背後で、古いバイナリをコピーしている。 古い方は非アクティブのままとしている。(トラフィックが発生しない?) 大規模障害に備えて、通常、新しいジョブを新しいバージョンのウェアハウスにルーティングしている ものを古いウェアハウスにルーティングするロールバックをできるようにしている。 Issueに基づいて顧客ごとに対象を絞ったロールバックをすることもあるらしい。 顧客のワークロードはそれぞれ大分ことなるので、全員が同じ頻度でバグを踏むことはないので。 特定の顧客に対して、アップデートした一部のリリースをロールバックする、みたいなことをするらしい。 リリースノートの扱いが良い感じになっていて、こういう感じで運用されているのだな、と。 将来の機能 現在、ユーザは、ワークロードの複雑さ、処理時間、コストを考慮して適切な調整を行わないといけない。 例えば、サイズ、ウェアハウスタイプ、クラスタ数、スケーリングポリシーなど。 こういったキャパシティ調整の大変さを減らしたり無くそうとしているらしい。 microVM(例えばFirecrackerやKata Containersなど)やシステムコールのオフロードに 投資し、より強力なサンドボックス分離メカニズムを実現しようとしているらしい。 それにより、Python/Javaコードで現状ではできないことが出来るようにしたいらしい。 まとめ Deep dive into the internals of Snowflake Virtual Warehousesを読んでみました。 たぶん公開されていない内部の仕組みの割合が多いのかなと思いましたがどうでしょうか。 正直かなり難しくて、途中、ほとんど写経状態になっている部分もありますが、 なるべく分からないところを調べながら、何を言いたいのかを趣旨の理解に努めました。 正直、知らなくても問題ないし、公開されていない以上、実際は違うかもしれないし、 将来変更されてしまうかもしれません。 1週間ぐらいかけて読んでみて、公開されている仕様を説明しやすくなった気はしました。

SnowflakeはS3, Blob, GCSを外部ステージとして設定し、データをロードする機能を備えています。 Snowflakeは各クラウドストレージとの接続方法として「ストレージ統合」の使用を推奨しています。 今回、GCSとのストレージ統合を設定し、外部ステージのロード先としてみました。 その手順等を感想を混ぜつつ書いてみようと思います。 参考にしたSnowflake公式ドキュメントは以下です。 ストレージ統合の設定方法がステップbyステップで示されています。 Google Cloud Storageの統合の構成 この記事は、公式が説明するステップに従って書いていきます。 [arst_toc tag=\"h4\"] GCSとストレージ統合 一般に、各ストレージサービスと連携する際、接続に必要な接続情報が必要です。 接続情報を自力で保管し使用する場合、セキュリティリスクに晒されるため、 Snowflakeはよりセキュアな方法として「ストレージ統合」機能を提供しています。 現在、Snowflakeは「ストレージ統合」機能のみを推奨しています。 各クラウドプロバイダ毎に実装方法は異なり、GCSの場合、Service Accountを使用します。 クラウドプロバイダ間で概念が異なるため 「AWSでいうところのXXXだよね」 は誤解を生みますが、 簡単に言えば、AWSのIAM、Azureのサービスプリンシパル、が相当すると思います。(本当??) 以下の図はその概念図です。(公式直リンク) Snowflakeは、ストレージ統合オブジェクトをGCS用のサービスアカウントと紐付けます。 このサービスアカウントは裏でSnowflakeが作成します。 言い換えると、GCSの認証責任をSnowflakeが作成するサービスアカウントに委任します。 Snowflake上のロード・アンロード操作の裏で透過的にサービスアカウントの権限参照が行われます。 Snowflakeでストレージ統合を作成する Snowflake上でストレージ統合を作る方法は以下の通りです。 実行にはACCOUNTADMINロールか、CREATE INTEGRATION権限が必要です。 ストレージ統合を作成する段階では、GCP側の権限設定は不要です。 CREATE STORAGE INTEGRATION gcs_int TYPE = EXTERNAL_STAGE STORAGE_PROVIDER = \'GCS\' ENABLED = TRUE STORAGE_ALLOWED_LOCATIONS = (\'gcs://mybucket1/path1/\', \'gcs://mybucket2/path2/\') STORAGE_BLOCKED_LOCATIONS = (\'gcs://mybucket1/path1/sensitivedata/\', \'gcs://mybucket2/path2/sensitivedata/\'); 自動生成されたサービスアカウントの情報を取得 CREATE INTEGRATIONによって作られたSnowflake用のサービスアカウント情報を取得します。 オブジェクトの詳細情報を取得する DESC コマンドにより取得できます。 DESC STORAGE INTEGRATION gcs_int; +-----------------------------+---------------+-----------------------------------------------------------------------------+------------------+ | property | property_type | property_value | property_default | +-----------------------------+---------------+-----------------------------------------------------------------------------+------------------| | ENABLED | Boolean | true | false | | STORAGE_ALLOWED_LOCATIONS | List | gcs://mybucket1/path1/,gcs://mybucket2/path2/ | [] | | STORAGE_BLOCKED_LOCATIONS | List | gcs://mybucket1/path1/sensitivedata/,gcs://mybucket2/path2/sensitivedata/ | [] | | STORAGE_GCP_SERVICE_ACCOUNT | String | service-account-id@project1-123456.iam.gserviceaccount.com | | +-----------------------------+---------------+-----------------------------------------------------------------------------+------------------+ STORAGE_GCP_SERVICE_ACCOUNTプロパティが、作成されたサービスアカウントです。 バケットオブジェクトにアクセスするためのサービスアカウント権限を付与する 最後に、サービスアカウントがバケットへアクセスするための権限設定を行います。 ここで初めて、GCPコンソール上での操作が必要となります。 カスタムIAMロールの作成 以下の権限を持つIAMロールを作成します。今回はロードのみを行うため以下を設定します。 パージを行う場合はobjects.delte、アンロードを行う場合はobjects.createが必要です。 storage.buckets.get storage.objects.delete storage.objects.get storage.objects.list サービスアカウントへIAMロールを割り当てる GCPのやり方に従って、サービスアカウントへIAMロールを割り当てます。 バケットに対して、サービスアカウントのアクセスを許可し、その許可内容をIAMロールで指定します。 Azureと比較するとAWSとGCPは近いなと感じます。 外部ステージの作成 まず、ステージオブジェクトを格納するSnowflakeデータベースへの権限付与を行います。 GRANT USAGE ON DATABASE mydb TO ROLE myrole; GRANT USAGE ON SCHEMA mydb.stages TO ROLE myrole; GRANT CREATE STAGE ON SCHEMA mydb.stages TO ROLE myrole; GRANT USAGE ON INTEGRATION gcs_int TO ROLE myrole; 次にストレージ統合を指定して外部ステージオブジェクトを作成します。 USE SCHEMA mydb.stages; CREATE STAGE my_gcs_stage URL = \'gcs://mybucket1\' STORAGE_INTEGRATION = gcs_int FILE_FORMAT = my_csv_format; 最後に疎通試験を行います。 copy into testtable from @my_gcs_stage pattern=\'testdata.csv\'; まとめ Snowflakeの公式ドキュメントの通りにGCSとのストレージ統合を作成しました。 また、作成したストレージ統合上に外部ステージを設定し、ロードが出来ることを確認しました。

Azure側を最小権限で構成する方法が書かれた記事がありそうで全然見つからない。 そこで、本記事ではAzure側を最小権限で構成する方法を調べてまとめてみた。 2023年7月現在、公式はSASを使う方法ではなくストレージ統合を使う方法を推奨している。 ストレージ統合によりAzureADへ認証を委譲することで認証情報を持ち回る必要がなくなる。 本記事ではストレージ統合を使用する。 ストレージアカウントにはURL(認証)・N/Wの2系統のパブリックアクセスの経路が存在するが、 URL(認証)によるパブリックアクセスは遮断し、Snowflake VNetからのアクセスのみ許可する。 Azure Portal/CLIからのアクセスを固定IPで制限する。 [arst_toc tag=\"h4\"] ストレージアカウント ストレージアカウントは、Blob(Object)、Files(SMB)、Queue、Table(NoSQL)の4種類が存在する。 Snowflakeの外部ステージとして使用するのはAWS S3でありAzure Blobである。 Blobは最上位階層にない。ストレージアカウント内にはコンテナがあり、コンテナ内にBlobがある。 Blobストレージ内のオブジェクトに対してのみ、外部からアクセス(パブリックアクセス)できる。 パブリックアクセスには、URL・ネットワークの2経路が存在し、それぞれ独立して存在する。 他方を制限したからといってもう一方が自動的に制限されたりはしない。 パブリックURLアクセス ストレージアカウントレベルで、パブリックアクセスを以下から選択できる。 Enabled Disabled また、コンテナレベルで、パブリックアクセスを以下から選択できる。 Private (no anonymous access) Blob (anonymous read access for blobs only) Container (anonymous read access for containers and blobs) 上位階層であるストレージアカウントレベルの設定が優先される。上位で不許可にすれば不許可。 もしストレージアカウントレベルで許可した場合、コンテナレベルの設定が効く。 その組み合わせは以下の通り。確かにMECEなのだが絶望的に冗長に思える。 コンテナ自体はURLアクセス不許可だがBlobはURLアクセス可とか設定できてしまう。 パブリックネットワークアクセス パブリックネットワークアクセスとして以下から選択できる。 Enabled from all network Enabled from selected virtual networks and IP addresses Disabled 分かりづらいが、ストレージアカウント/Blobへのアクセスに使われるN/Wインターフェースを選ぶ。 Disabledにすると、仮想ネットワークのプライベートIPアドレスを使用するN/Wインターフェースとなる。 Enabledには2つあり、全てのN/Wからアクセス可能なN/Wインターフェースか、 選択した仮想N/WまたはIPアドレスからアクセス可能なN/Wインターフェースを選ぶ もちろんDisabledにした場合は、仮想プライベートエンドポイントが必要となる。 ストレージ統合の際のストレージアカウントの最小権限構成 以下のようにして最小権限構成を行なった。 パブリックURLアクセスをストレージアカウントレベルでDisabled パブリックネットワークアクセスを固定IP、及びSnowflake VNet IDsに設定 Snowflakeのストレージ統合機能によりサービスプリンシパルを作成 ストレージコンテナのIAMでサービスプリンシパルにBlob読み書きロール設定 ストレージアカウントに対するSnowflakeのVNetサブネットIDsからのアクセス許可 ストレージアカウントへのパブリックアクセスを「選択したVNetまたはIPアドレスのみ」とした場合、 SnowflakeのVNetまたはIPアドレスが分かる必要がある。IPアドレスは不定なのでVNet IDsを指定する。 公式にドンピシャの説明があるので、それを参考にVNet IDsを設定する。 VNet サブネット IDs の許可 Snowflake側で以下のSQLを実行すると、属するVNet IDがJSONで複数出力される。 それを記録しておく。 USE ROLE ACCOUNTADMIN; SELECT SYSTEM$GET_SNOWFLAKE_PLATFORM_INFO() ; { \"snowflake-vnet-subnet-id\":[ \"/subscriptions/hogehoge\", \"/subscriptions/fugafuga\" ] } 続いて、ストレージアカウントに上の2個のVNet IDからのアクセスを許可する。 以下のように、Azure CLIでログインしてVNet IDの個数だけコマンドを実行する。 $ az storage account network-rule add --account-name --resource-group --subnet \"/subscriptions/hogehoge\" ... $ az storage account network-rule add --account-name --resource-group --subnet \"/subscriptions/fugafuga\" ... ストレージ統合を作成する AWS環境と同様に、CREATE STORAGE INTEGRATION によりストレージ統合を作成する。 Azure CLIやAzure PortalでテナントIDを取得しておいてAZURE_TENANT_IDに渡す。 STORAGE_ALLOWED_LOCATIONSにはBLOBの位置を表す文字列を渡す。 書式は azure://ストレージアカウント名.blob.core.windows.net/コンテナ名/パス 。 BLOBの位置は複数渡せる。 CREATE STORAGE INTEGRATION storage_integration_azure TYPE = EXTERNAL_STAGE STORAGE_PROVIDER = \'AZURE\' ENABLED = TRUE AZURE_TENANT_ID = \'\' STORAGE_ALLOWED_LOCATIONS = (\'azure://storageikuty1.blob.core.windows.net/container1/files/\') ; 続いて、同意URL(AZURE_CONSENT_URL)を取得する。 DESC STORAGE INTEGRATION storage_integration_azure ; property property_type property_value property_default ENABLED Boolean true false STORAGE_PROVIDER String AZURE STORAGE_ALLOWED_LOCATIONS List azure://storageikuty1.blob.core.windows.net/container1/files/ [] STORAGE_BLOCKED_LOCATIONS List [] AZURE_TENANT_ID String AZURE_CONSENT_URL String https://login.microsoftonline.com/hogehoge&response_type=code AZURE_MULTI_TENANT_APP_NAME String COMMENT String URLを開き、表示された画面で「承認」ボタンを押下する。 この際、権限がなければ権限不足である旨表示されてここで終了する。(権限は後述) このタイミングでAzure環境にSnowflakeと対応するサービスプリンシパルが作られる。 上手くいくとSnowflakeの企業サイトに遷移する。 Azure Portalにログインし、Snowflakeと共有したいストレージアカウントを選択する。 IAMから、サービスプリンシパルに対して以下のロールを割り当てる。 サービスプリンシパルIDは、上で得られたマルチテナントアプリ名のハイフンの前の部分。 この辺り、むちゃくちゃ煩雑で、AWSよりも大変。 Storage Blob Data Reader(ストレージBlobデータ閲覧者) Storage Blob Data Contributor(ストレージBlobデータ共同作成者) 次にDBにフォーマットと(外部)ステージを作成する。 USE ROLE SYSADMIN; USE DATABASE HOGE; CREATE OR REPLACE FILE FORMAT MY_CSV_FORMAT TYPE = CSV COMMENT = \'接続検証\' ; CREATE OR REPLACE STAGE my_azure_stage STORAGE_INTEGRATION = storage_integration_azure URL = \'azure://storageikuty1.blob.core.windows.net/container1/files/\' FILE_FORMAT = my_csv_format; ; 該当のBlobにファイルをアップロードする。(パスを指定した場合忘れがち) そして疎通確認。 $ show storage_integration_azure ... $ list @storage_integration_azure ... まとめ Azure環境でストレージ統合によりBlobアクセスを設定してみた。 その際、Azure環境側でパブリックアクセスの開き方が最小限になるように構成した。 Azure側、Snowflake側共に大きな権限が必要だが、公式推奨だし可能ならこの方法が良さそう。

ドキュメントを読んで使ったことが無い機能に詳しくなるシリーズ。 今回は検索最適化サービス(Search Optimization Service)。 他のドキュメントと比較して英語・日本語ともに文体が異なる気がする。 日本語訳が変なのではなく、元の英文の文体が違う。 まず、\"クラスタリング\"とか\"Materialized View\"が構造の名称である一方で、 \"検索最適化\"が目的の名称となっている点に違和感を感じる。 似て非なる索引(index)とは違うのだよ、という何かなのだろうか... [arst_toc tag=\"h4\"] 検索最適化サービスの概要 検索最適化サービスは、1つまたは少数の異なる行のみを返す「ポイントルックアップクエリ」。むむ?。 要は、巨大なテーブルに対するSELECT文の結果が数行となるようなケースを高速化する。 例えば、あるテーブルのageというカラムに対する等価条件を設定したクエリを速くしたい場合、 あらかじめ高速化用のデータ構造を構築し、ageに対する等価条件を使ったクエリを高速化する。 等価条件検索、範囲条件検索、部分文字列検索に加え、 プレビュー機能として、正規表現検索、半構造化型値検索、地理空間関数検索 が用意される。 他の高速化技術との比較 高速化機能として、他にクラスタリング、Materialized Viewがあるが、それぞれ機序が異なる。 クラスタリングは、マイクロパーティションの分散と枝刈りを利用した方法であって、 ナチュラルキーよりも高いプルーニング効率が得られるケースに適用する。 Materialized Viewは、同一テーブルに対して複数のクラスタリングキーを透過的に適用する 用途で用いられる。もちろん直アクセスでは遅い場所の事実上のキャッシュとしても使う。 検索最適化サービスは、ある意味、RDBMSの索引チックな使い方ができる。 良く使うカラムと検索述語を「アクセスキー」として登録し、高速検索用の索引を構築する。 検索最適化サービスの構成に必要な権限 以下の権限が必要。 テーブルの所有権 テーブルを含むスキーマに対する ADD SEARCH OPTIMIZATION 権限 検索最適化サービスを適用したテーブルにアクセスするには普通にSELECTできれば良い。 検索最適化サービスで未サポートのもの 以下に対して検索最適化サービスを適用することはできない。 (逆に以下以外なら適用できる。) 外部テーブル マテリアライズドビュー COLLATE 句 で定義された列 列の連結 分析表現 テーブル列にキャスト（文字列にキャストされる固定小数点数を除く） Materialized Viewは外部テーブルアクセスを高速化できるが、 検索最適化は外部テーブル、Materialized Viewを対象に出来ない。 何かの事情でCOLLATE句を指定した場合、その列は使用できない。 述語を指定する際に、定数側の明示的キャストはサポートされるが、 列側をキャストすると効かなくなる。例えば以下のような違いがある。 ただし、NUMBERからVARCHARへのキャストだけは機能する。 -- 機能する SELECT hoge FROM table as t1 WHERE timestamp1 = \'2020-01-01\'::date; -- 機能する SELECT hoge FROM table as t1 WHERE cast(numeric_column as varchar) = \'2\' -- 機能しない SELECT hoge FROM table as t1 WHERE to_date(varchar_column) = \'2020-01-01\'; (分析表現って何!?) 検索最適化サービスの構成方法 ALTER TABLE と ADD SEARCH OPTIMIZATION を使用する。 テーブル全体(つまり全カラム)に対して適用できるか、カラムを指定することもできる。 カラムを指定する場合、ON句に最適化したい検索述語等を指定する。 例えば以下であればc1,c2列を使用した等価述語が最適化される。 c1列、c2列に対して等価述語(=)を使ったSQLが高速化される。 -- t1テーブルの c1,c2 の等価述語に対して検索最適化を適用 ALTER TABLE t1 ADD SEARCH OPTIMIZATION ON EQUALITY(c1,c2); -- クエリ SELECT hoge FROM t1 where c1 = 100 and c2 = 200; 検索最適化サービスが適用済みか確認する方法 DESCRIBE により該当テーブルへの検索最適化サービスの適用状況を確認できる。 定義した「検索アクセスパス」の詳細と、有効無効が表示される。 DESCRIBE SEARCH OPTIMIZATION ON t1; +---------------+----------+--------+------------------+--------+ | expression_id | method | target | target_data_type | active | +---------------+----------+--------+------------------+--------+ | 1 | EQUALITY | C1 | NUMBER(38,0) | true | +---------------+----------+--------+------------------+--------+ テーブル全体に対して検索最適化サービスを適用していた場合、 列の追加により、追加した列に対して自動的に検索最適化サービスが適用される。 逆に列の削除により、削除した列が検索最適化サービスから除外される。 列のデフォルト値を無効にすると、検索最適化サービスが「無効」になる。 一度「無効」になってしまったら、SEARCH OPTIMIZATION を一度 DROP し、 再度 ADD しなければならない。 検索最適化サービスのコスト 以下のコストがかかる。 ストレージリソース コンピューティングリソース ストレージリソース 検索アクセスパスのためのデータ構造を構築するために使われる。 通常、元テーブルの1/4ぐらいのサイズになるらしい。 ただし、テーブル内の個別の値の数が多ければ多いほどストレージを消費する。 例えば、全カラムを対象としている場合、全カラムの値が全て異なるという状況は最悪で、 元テーブルと同じだけのストレージを消費してしまう。 コンピューティングリソース 検索最適化を維持するためにコンピューティングリソースが使われる。 テーブル内で大量のデータが変更される場合、リソースの消費量が激しくなる。 追加・変更に比例して大きくなる。削除は若干使用する。 ざっくり以下に比例してコストが増加する。 この機能が有効になっているテーブルの数、およびそれらのテーブル内の個別の値の数。 これらのテーブルで変更されるデータの量 SYSTEM$ESTIMATE_SEARCH_OPTIMIZATION_COSTS 関数によりコストを見積もれる。 -- テーブルに検索最適化を追加する場合の推定コスト select SYSTEM$ESTIMATE_SEARCH_OPTIMIZATION_COSTS(\'TABLE_WITHOUT_SEARCH_OPT\'); +---------------------------------------------------------------------------+ | SYSTEM$ESTIMATE_SEARCH_OPTIMIZATION_COSTS(\'TABLE_WITHOUT_SEARCH_OPT\') | |---------------------------------------------------------------------------| | { | | \"tableName\" : \"TABLE_WITHOUT_SEARCH_OPT\", | | \"searchOptimizationEnabled\" : false, | | \"costPositions\" : [ { | | \"name\" : \"BuildCosts\", | | \"costs\" : { | | \"value\" : 11.279, | | \"unit\" : \"Credits\" | | }, | | \"computationMethod\" : \"Estimated\", | | \"comment\" : \"estimated via sampling\" | | }, { | | \"name\" : \"StorageCosts\", | | \"costs\" : { | | \"value\" : 0.070493, | | \"unit\" : \"TB\" | | }, | | \"computationMethod\" : \"Estimated\", | | \"comment\" : \"estimated via sampling\" | | }, { | | \"name\" : \"MaintenanceCosts\", | | \"costs\" : { | | \"value\" : 30.296, | | \"unit\" : \"Credits\", | | \"perTimeUnit\" : \"MONTH\" | | }, | | \"computationMethod\" : \"Estimated\", | | \"comment\" : \"Estimated from historic change rate over last ~11 days.\" | | } ] | | } | +---------------------------------------------------------------------------+ -- すでに検索最適化が行われているテーブルに対するこの関数の出力 select SYSTEM$ESTIMATE_SEARCH_OPTIMIZATION_COSTS(\'TABLE_WITH_SEARCH_OPT\'); +---------------------------------------------------------------------------+ | SYSTEM$ESTIMATE_SEARCH_OPTIMIZATION_COSTS(\'TABLE_WITH_SEARCH_OPT\') | |---------------------------------------------------------------------------| | { | | \"tableName\" : \"TABLE_WITH_SEARCH_OPT\", | | \"searchOptimizationEnabled\" : true, | | \"costPositions\" : [ { | | \"name\" : \"BuildCosts\", | | \"computationMethod\" : \"NotAvailable\", | | \"comment\" : \"Search optimization is already enabled.\" | | }, { | | \"name\" : \"StorageCosts\", | | \"costs\" : { | | \"value\" : 0.052048, | | \"unit\" : \"TB\" | | }, | | \"computationMethod\" : \"Measured\" | | }, { | | \"name\" : \"Benefit\", | | \"computationMethod\" : \"NotAvailable\", | | \"comment\" : \"Currently not supported.\" | | }, { | | \"name\" : \"MaintenanceCosts\", | | \"costs\" : { | | \"value\" : 30.248, | | \"unit\" : \"Credits\", | | \"perTimeUnit\" : \"MONTH\" | | }, | | \"computationMethod\" : \"EstimatedUpperBound\", | | \"comment\" : \"Estimated from historic change rate over last ~11 days.\" | | } ] | | } | +---------------------------------------------------------------------------+ まとめ 検索最適化サービスについてドキュメントを読んでいくつか試してみた。 ドキュメントにある「ポイントルックアップクエリ」という言葉が機能を最も適切に表していると思う。 高速化したいクエリが明確な時に使えるのかもしれない。

使ったことがない機能のドキュメントを読んで詳しくなるシリーズ。 今回は外部関数(External Function)。 [arst_toc tag=\"h4\"] ユーザ定義関数 まずユーザ定義関数(User Defined Function)について。 UDFは、組み込み関数と同様にSQLから呼び出すことができる。組み込み関数にない機能を自作できる。 コードを共通化して2個以上の場所で発生するコードクローンを除去できる。 ユーザ定義関数は単一の値を返すパターンと、表を返すパターンの2パターンを定義できる。 単一の値を返すパターンはスカラー関数(UDFs)。 表、つまり0個,1個,2個以上の値を返すパターンは表関数(UDTFs)。 サポートされている言語はJava,JavaScript,Python,SQL。 スカラー関数(UDFs) スカラー関数は入力行に対して1つの出力行を返す。 公式のサンプルを読むのが早い。 -- UDFの定義 create or replace function addone(i int) returns int language python runtime_version = \'3.8\' handler = \'addone_py\' as $$ def addone_py(i): return i+1 $$; -- UDFの実行 select addone(1); 2 こんな調子で、create function の中に書き下した処理をインラインハンドラと言う。 処理を修正する際は、alter function を実行して対応する。 確かにお手軽だが、行数が増えてきたり、処理を共通化したりしたい場合は微妙。 あらかじめコードをコンパイルしておき再利用することもできる(ステージングハンドラ)。 ステージングハンドラであれば、より多くの行数に対応できるし、処理の共通化も用意。 ステージングハンドラの実行例は以下の通り。 CREATE FUNCTION my_udf(i NUMBER) RETURNS NUMBER LANGUAGE JAVA IMPORTS = (\'@mystage/handlers/my_handler.jar\') HANDLER = \'MyClass.myFunction\' インラインハンドラは全ての言語で対応できるが、 ステージングハンドラは、JavaまたはPythonが必要。 表関数(UDTFs) 表関数は入力行に対して表、つまり0行,1行,2行,...行 を返す。 実行例は以下の通り。なるほど。 戻り値が表(table)になるため、SQLだと処理と戻り値のギャップが少ないが、 SQLでは手続き的な処理が書けないため自由度は低い。 -- サンプルテーブル・データの作成 create or replace table orders ( product_id varchar, quantity_sold numeric(11, 2) ); insert into orders (product_id, quantity_sold) values (\'compostable bags\', 2000), (\'re-usable cups\', 1000); -- UDTFの定義 create or replace function orders_for_product(PROD_ID varchar) returns table (Product_ID varchar, Quantity_Sold numeric(11, 2)) as $$ select product_ID, quantity_sold from orders where product_ID = PROD_ID $$ ; -- UDTFの実行 select product_id, quantity_sold from table(orders_for_product(\'compostable bags\')) order by product_id; +------------------+---------------+ | PRODUCT_ID | QUANTITY_SOLD | |------------------+---------------| | compostable bags | 2000.00 | +------------------+---------------+ JavaScriptで書くと以下のようになる。手続き的に処理を書いている。 これであればある程度の自由度を獲得できる。 -- UDTFの定義 -- IP アドレスの「範囲」を IP アドレスの完全なリストに変換 CREATE OR REPLACE FUNCTION range_to_values(PREFIX VARCHAR, RANGE_START FLOAT, RANGE_END FLOAT) RETURNS TABLE (IP_ADDRESS VARCHAR) LANGUAGE JAVASCRIPT AS $$ { processRow: function f(row, rowWriter, context) { var suffix = row.RANGE_START; while (suffix <= row.RANGE_END) { rowWriter.writeRow( {IP_ADDRESS: row.PREFIX + "." + suffix} ); suffix = suffix + 1; } } } $$; -- UDTFの実行 SELECT * FROM TABLE(range_to_values('192.168.1', 42::FLOAT, 45::FLOAT)); +--------------+ | IP_ADDRESS | +==============+ | 192.168.1.42 | +--------------+ | 192.168.1.43 | +--------------+ | 192.168.1.44 | +--------------+ | 192.168.1.45 | +--------------+ 外部関数 ユーザ定義関数のロジック(ハンドラ)を外部のコンピューティングリソースに配置することができる。 概念図は以下の通り。Snowflake側から見て、外部関数、API統合、プロキシサービス、リモートサービスから成る。 外部関数(External Function)はスカラー関数(UDFs)だが、自前でコードを書かない。 その実体は、プロキシサービスを介してリモートサービスのコードを実行するためのオブジェクトで、 CREATE EXTERNAL FUNCTION により定義する。 リモートサービスは、外部のコンピューティングリソース。 AWS Lambda、Azure Functionsが分かりやすいが、 HTTPSでJSONを受けてJSONを返せれば、一般的なWebサーバでも良い様子。 プロキシサービスは、要はリモートサービスを抽象化するレイヤの機能を持つ。 AWS API Gateway、Azure API Managementサービスと言うと分かりやすい。 外部関数が直接リモートサービスを叩かない理由は、プロキシサービスに認証機能を持たせるため。 API統合は、外部関数をSnowflakeの世界に持ち込むための抽象化レイヤ。 同様の機能として、S3,Blobへのアクセス手段を抽象化するストレージ統合がある。 使い場合はUDFと同様。 select my_external_function_2(column_1, column_2) from table_1; select col1 from table_1 where my_external_function_3(col2) < 0; create view view1 (col1) as select my_external_function_5(col1) from table9; select upper(zipcode_to_city_external_function(zipcode)) from address_table; リモートサービスの入出力定義 前述のように、リモートサービスはHTTPSでJSONを受けてJSONを返す必要がある。 そのインターフェースは公式に記述がある。 リモートサービスの入力および出力データ形式 基本的に、POSTを受けて同期的に応答するようにする。 ただし、非同期処理もサポートしており、その際はGETでポーリングリクエストを受ける必要がある。 関数 ext_func(n integer) returns varchar のケースにおいて、リクエストヘッダは以下。 いわゆるRPC(Remote Procedure Call)を定義する何かに準じている。 ヘッダ文字列値 sf-external-function-formatjson固定 sf-external-function-format-version1.0固定 sf-external-function-query-batch-idリクエスト固有の識別子。この値を使って処理の成否を確認する。 sf-external-function-name関数名 ext_func sf-external-function-name-base64関数名(base64) base64(ext_func) sf-external-function-signature関数名署名 (N NUMBER) sf-external-function-signature-base64関数名署名(base64) base64(N NUMBER) sf-external-function-return-type戻り値型 VARCHAR(16777216) sf-external-function-return-type-base64戻り値型(base64) base64(VARCHAR(16777216)) リクエスト本文はJSON。 重要なことは、外部関数はスカラー関数であり、1個の入力に対して1個の応答を返す必要があるが、 一度に複数の入力を受けることができ、それぞれに対して応答を返す必要がある、ということ。 複数の入力のセットを公式ではバッチ、と読んでいる様子。 署名 f(integer, varchar, timestamp) を持つ外部関数にシリアル化したJSONを送る。例は以下。 先頭の数字が1個多い気がするが、これは行番号を表す。 { \"data\": [ [0, 10, \"Alex\", \"2014-01-01 16:00:00\"], [1, 20, \"Steve\", \"2015-01-01 16:00:00\"], [2, 30, \"Alice\", \"2016-01-01 16:00:00\"], [3, 40, \"Adrian\", \"2017-01-01 16:00:00\"] ] } リモートサービスの応答定義 レスポンス本文はJSON。リクエスト本文と同様にコレクションの先頭は行番号を設定する必要がある。 公式には、都市名を引数として、その緯度軽度を返す関数の例が書いてある。 このように、入力と対応する応答値を、入力と同じ順番にコレクションに詰めてシリアライズして返す。 この外部関数は、VARIANT型を戻している。このように、スカラー関数ではあるが、 1回の処理が(半)構造を持てるので、事実上複数の入力値に対して複数の応答値を返せる。 -- 呼び出し例 select val:city, val:latitude, val:longitude from (select ext_func_city_lat_long(city_name) as val from table_of_city_names); -- レスポンス { \"data\": [ [ 0, { \"City\" : \"Warsaw\", \"latitude\" : 52.23, \"longitude\" : 21.01 } ], [ 1, { \"City\" : \"Toronto\", \"latitude\" : 43.65, \"longitude\" : -79.38 } ] ] } StatusCode=200で、同期応答完了。 202を返すと、非同期応答となり処理を継続できる様子だが今回は省略。 その他はエラー応答として扱われる。 まとめ 外部関数に関するドキュメントと、その理解に必要なUDFs、UDTSsのドキュメントを読んでみた。 実体はHTTPS上のWebAPIであり、事実上無限の自由度を獲得できると思う。 (これを持ち出す頃には、本末転倒な何かに陥っていないか注意は必要だと思う)