Snowflake×Streamlit×PyGWalkerの始め方

この記事は[Snowflake Advent Calendar 2023](https://qiita.com/advent-calendar/2023/snowflake)の1日目です。
今年も来年もみんなでSnowflakeを盛り上げよう！

# PyGWalkerとは

PyGWalkerはPandasのDataFrameをTableau風のGUIで探索・可視化できるPythonライブラリです。  
オープンソースソフトウェアとして開発されていてソースコードはGitHubで公開されています。  

なお、[README](https://github.com/Kanaries/pygwalker/blob/main/docs/README.ja.md)によると、
> "Pig Walker"のように発音  

…だそうです。

PyGWalkerを起動させると以下のようなWebアプリが立ち上がります！  

![Example of PyGWalker](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2917580/5276374b-227e-7ea4-b41e-71e38841a391.png)

## PyGWalkerをSnowflakeと組み合わせると何がいいのか

:::note warn
この記事に記載したプログラムは、ローカルPCなどで動作しているStreamlitからSnowflakeに接続しています。
Streamlit in Snowflakeでは動作しないのでご注意ください。  
:::

PyGWalkerはデフォルトではすべての処理をフロントエンド(つまりブラウザ上)でやろうとします。そのため、データが大きくなるとブラウザで処理しきれなくなってしまいます。  
これを解決するため、PyGWalkerにはバックエンドに集計やフィルターなどの処理を任せる(プッシュダウン)ことができます。  
さらに、バックエンドとしてSnowflakeやBigQueryを選択することで、クラウドならではの膨大なリソースを活用することができます！大規模なデータを探索・可視化するのにピッタリですね。  

以下がイメージ図です。
![Image of Pushdown](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2917580/2c622142-cb3f-71a4-c535-87d245a81ddb.png)

## 環境準備

[公式ドキュメント](https://docs.kanaries.net/ja/pygwalker/installation)を参考にしながらインストールしていきます。  

https://docs.kanaries.net/ja/pygwalker/installation

まずは必要なパッケージをインストールします。  
自分はAnaconda環境で構築したため、ところどころに```conda```コマンドが混じっています。  

```sh
conda create -n pygwalker python=3.10
conda activate pygwalker
pip install pygwalker
pip install streamlit
pip install snowflake-sqlalchemy
```

また、今回の接続方法だとSnowflakeはデフォルトウェアハウスでクエリを実行しようとします。  
そのため、あらかじめユーザに対してデフォルトウェアハウスを設定しておきましょう。  
参考文献1：[ユーザーのデフォルトウェアハウス](https://docs.snowflake.com/ja/user-guide/warehouses-overview#default-warehouse-for-users)  
参考文献2：[Snowflakeのデフォルトウェアハウスは自分のものしか変更できない](https://qiita.com/foursue/items/227cfafd5b87db0b8921)  

```sql
ALTER USER PYGUSER SET DEFAULT_WAREHOUSE='COMPUTE_WH';
```

## ソースコードと解説

実際のソースコードは以下のようになります！  

```py:sample_with_snowflake.py
import streamlit as st
import streamlit.components.v1 as components
from pygwalker.api.streamlit import init_streamlit_comm, get_streamlit_html
from pygwalker.data_parsers.database_parser import Connector
import pandas as pd

# Streamlitのページ設定。タイトルとレイアウトを設定する
st.set_page_config(
    page_title="Snowflake X Streamlit X PyGWalker",
    layout="wide"
)
 
# 1.PyGWalkerの通信機能を初期化
init_streamlit_comm()

# 2.Snowflakeと接続するコネクタを取得
# 各パラメータをsecrets.tomlから取得する
username = st.secrets["snowflake_username"]
password = st.secrets["snowflake_password"]
account_identifier = st.secrets["snowflake_account_identifier"]
database = 'SNOWFLAKE_SAMPLE_DATA'
schema = 'TPCH_SF1'
# Connectorを作成する。SQLAlchemy形式のURLを指定する
url = f'snowflake://{username}:{password}@{account_identifier}/{database}/{schema}'
# このSQLの実行結果がPyGWalkerで探索できるようになる
sql = """
    SELECT
        *
    FROM
        SNOWFLAKE_SAMPLE_DATA.TPCH_SF1.ORDERS
"""
conn = Connector(url, sql)

# 3.PyGWalkerのフロントエンドのHTMLを取得する
# `use_kernel_calc=True`を指定するときは、PyGWalkerのHTMLをキャッシュすることが推奨
# そのため、`@st.cache_resource`を使用してキャッシュするが、このときハッシュできない引数 Connector の変数名には`_`をつけている
@st.cache_resource
def get_html(_conn: Connector) -> str:
    # Streamlitのコンポーネントに埋め込むHTMLを取得するときはwalk()の代わりにget_streamlit_html()を使用する
    # Snowflakeをバックエンドとして使用するときはDataFrameではなくConnectorを渡す
    # PyGWalkerの通信機能を使用するため、`use_kernel_calc=True`を指定する
    html = get_streamlit_html(_conn, use_kernel_calc=True, debug=False)
    return html

# 4.PyGWalkerのフロントエンドのHTMLをStreamlitのコンポーネントに埋め込む
html = get_html(conn) 
components.html(html, width=1300, height=1000, scrolling=True)
```

以下、解説です。  

PyGWalkerでバックエンドに集計やフィルターなどの処理を任せるときは、まず```init_streamlit_comm()```で通信機能を初期化する必要があります。  

```py
init_streamlit_comm()
```

そしてHTMLを取得するときに```use_kernel_calc=True```を指定しましょう。  
このとき、使用するメモリが肥大化しないように、Streamlitのキャッシュを使うことが推奨とのことです。  

```py
@st.cache_resource
def get_html(_conn: Connector) -> str:
    html = get_streamlit_html(_conn, use_kernel_calc=True, debug=False)
    return html
```

Streamlitのキャッシュは関数を呼び出すときの引数のハッシュ値を用いて制御されるのですが、ハッシュできない引数があるとエラーが発生します。  
このような引数には```_```で始まる変数名を付けてあげましょう。  
参考：[st.cache_resource](https://docs.streamlit.io/library/api-reference/performance/st.cache_resource)  
> Any parameter whose name begins with _ will not be hashed.

PyGWalkerでHTMLを取得する関数、```walk()```あるいは```get_streamlit_html()```には通常はPandasのDataFrameを渡すのですが、ここにはConnectorを渡すこともできます。  
ConnectorにはSQLAlchemy形式のURLと、実行するSQLを指定します。  
これにより、DataFrameの代わりに、URLで指定されたデータベースでSQLを実行した結果を探索できます。  

```py
url = f'snowflake://{username}:{password}@{account_identifier}/{database}/{schema}'
sql = """
    SELECT
        *
    FROM
        SNOWFLAKE_SAMPLE_DATA.TPCH_SF1.ORDERS
"""
conn = Connector(url, sql)
```

Connectorに設定するURLのホスト名(前述のソースコードで```account_identifier```という変数にした部分)には、SnowflakeのアカウントURLの一部分を設定します。  
具体的には、Snowsightで```アカウントURLをコピー```したとき、```https://xx12345.ap-northeast-1.aws.snowflakecomputing.com```のようなURLがクリップボードにコピーされるので、ここから```xx12345.ap-northeast-1.aws```の部分だけを抜き出したものを設定します。  

![Copy account URL](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2917580/3dbae755-9429-1d79-7c75-3b6992022f1b.png)

詳細は[Snowflake SQLAlchemyに関するドキュメント](https://docs.snowflake.com/ja/developer-guide/python-connector/sqlalchemy)を参照してください。  

https://docs.snowflake.com/ja/developer-guide/python-connector/sqlalchemy

また、```st.secrets```については、[Streamlitのドキュメント](https://docs.streamlit.io/streamlit-community-cloud/deploy-your-app/secrets-management)を参照してください。  
ユーザ名やパスワードのようなハードコーディングしたくない情報を外部のファイル(secrets.toml)から読み込むための機能です。  

https://docs.streamlit.io/streamlit-community-cloud/deploy-your-app/secrets-management

## 実行

ソースコードが完成したら実行しましょう。  
今回はStreamlitでホストするので、```streamlit run```します。

```sh
streamlit run sample_with_snowflake.py
```

デフォルトの設定のままであれば、Webアプリは```localhost:8501```で立ち上がります。  
ブラウザでアクセスしてみましょう。  

![Example of PyGWalker](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2917580/5276374b-227e-7ea4-b41e-71e38841a391.png)

## Snowflakeは何をしてくれているのか

前述の通り、PyGWalkerはSnowflakeに集計やフィルターなどの処理を任せることができます。  
このときSnowflakeが何をしてくれているのか確認してみましょう。  

まず、PyGWalkerで、日付範囲でフィルターして／日単位で行数を集計した結果をグラフ表示してみます。  

![Filter and Aggregation](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2917580/daf33dc7-ec09-dbf5-e480-4a1ba7e24416.png)

続いて、Snowsightで「アクティビティ」領域にある「クエリ履歴」を見てみましょう。  

![Snowsight Query History](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2917580/0aa3b616-5ffd-0d98-6884-7afe350fd246.png)

次のようなクエリが実行されていることが確認できました！  

```sql
SELECT "O_ORDERDATE", SUM(1) AS "gw_count_fid_sum"
FROM (SELECT * FROM SNOWFLAKE_SAMPLE_DATA.TPCH_SF1.ORDERS) AS temp_view_name
WHERE CAST(EXTRACT(epoch_second FROM CAST(CAST("O_ORDERDATE" AS TIMESTAMPTZ) AS TIMESTAMPNTZ)) * 1000 AS BIGINT) BETWEEN 757350000000 AND 788886000000
GROUP BY "O_ORDERDATE"
```

癖がありますが、```WHERE```句で日付範囲をフィルターして、```GROUP BY```と```SUM(1)```で行数を集計していることが確認できますね！  
クエリの効率は少し怪しいですが、ともあれこれなら大規模なデータでもサクサク探索することができそうです。