Streamlit in Snowflake 用のログおよびトレース¶

Pythonのログ記録モジュール¶

Pythonの組み込み``logging``モジュールを使用して、アプリからログメッセージを出力します。次の例では、INFOレベル以上のメッセージを標準出力に書き込むロガーを構成しています。

import logging
import sys

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s: %(message)s",
    stream=sys.stdout,
)

LOGGER = logging.getLogger("my_app")

重大度が低いものから高いものの順に、Pythonには以下のログレベルがあります。

• DEBUG

• INFO

• WARNING

• ERROR

レベルをINFOに設定すると、INFO、WARNING、ERRORメッセージがキャプチャされますが、DEBUGメッセージはキャプチャされません。

ロガーを構成したら、アプリコード全体でメッセージをログに記録するために使用できます。ロガーは別のモジュールで定義し、アプリコードにインポートするのが一般的です。

source_directory/
├── my_logger.py
├── pyproject.toml
└── streamlit_app.py

import streamlit as st
from my_logger import LOGGER

LOGGER.info("Home page loaded")
st.title("My App")

if st.button("Run analysis"):
    LOGGER.info("Analysis button clicked")
    try:
        result = run_analysis()
        LOGGER.info("Analysis completed successfully")
    except Exception as e:
        LOGGER.error("Analysis failed: %s", e)
        st.error("Analysis failed: %s", e)

|sf-web-interface|のライブログ¶

|sf-web-interface|でコンテナランタイムアプリを編集する場合、エディターの下にログペインが表示されます。このペインには、アプリがログメッセージを出力すると、リアルタイムでログメッセージがストリーミングされます。最初に接続したときに、最新のログの簡単な履歴が表示されます。

各ログエントリは、次の情報を示します。

利用可能なライブログアクション¶

ログペインの右上隅で、ログを検索およびフィルターして、必要な情報を見つけることができます。これには、テキスト検索、ソースによるフィルタリング、重大度レベルによるフィルタリングが含まれます。3つのドットメニューで、現在のログをダウンロードしたり、過去のログに移動したり、ライブログペインをクリアしたりできます。ペインをクリアすると、現在のログは現在のビューから削除されますが、イベントテーブルからは削除されません。ページをすぐにリロードすると、最新のログが復元されます。

ログソースの理解¶

コンテナランタイムアプリからのログには、2つのソースのいずれかがあります。

• MANAGER:アプリを準備して実行するコンテナ内のシステムプロセス。マネージャーログには、ステージからのアプリファイルのダウンロード、Python依存関係のインストール、Streamlitサーバープロセスの開始に関するメッセージが含まれます。アプリの実行中にアプリの依存関係ファイルを更新すると、マネージャープロセスは依存関係を再インストールし、追加のマネージャーログを生成します。

• APP:実行中のStreamlitサーバープロセスからのログ。これには、ユーザー設定のPythonロガー、Streamlitの組み込みロガーからのメッセージ、およびアプリが標準出力や標準エラーに書き込むその他の出力が含まれます。

ソース間の境界は``streamlit run``コマンドです。Streamlitプロセスを開始する前にコンテナが行うすべてのタスクは、``MANAGER``ログを生成します。Streamlitプロセスが開始されると、そのプロセスからの出力は``APP``ログを生成します。

|sf-web-interface|での履歴ログの表示¶

次のステップは、コンテナランタイムアプリにのみ適用されます。ウェアハウスランタイムアプリにはログペインがありません。

1. Snowsight にサインインします。

2. ナビゲーションメニューで、 Projects » Streamlit を選択してから、アプリを選択します。

3. ページの右上隅で、 Edit を選択します。

4. ログペインの右上隅で、3つのドットメニュー（Other actions）|raa| :ui:`Historical logs`を選択します。

これにより、アプリの背後で実行されているサービスの|SPCS|モニタリングページが開きます。ログテーブルには、次の列が表示されます。

モニタリングページの詳細については、:doc:`/developer-guide/snowpark-container-services/monitoring-services`をご参照ください。

SQLでのログのクエリ¶

イベントテーブルを直接クエリして、コンテナランタイムアプリのログを分析できます。以下のクエリは、特定のStreamlitアプリからログを取得します。

SELECT
    TIMESTAMP,
    RECORD['severity_text']::VARCHAR AS level,
    VALUE::VARCHAR AS message,
    RESOURCE_ATTRIBUTES['snow.database.name']::VARCHAR AS database_name,
    RESOURCE_ATTRIBUTES['snow.schema.name']::VARCHAR AS schema_name,
    RESOURCE_ATTRIBUTES['snow.executable.name']::VARCHAR AS app_name,
    RECORD_ATTRIBUTES['log.iostream']::VARCHAR AS stream
FROM <event_table>
WHERE RESOURCE_ATTRIBUTES['snow.database.name'] = '<database_name>'
  AND RESOURCE_ATTRIBUTES['snow.schema.name'] = '<schema_name>'
  AND RESOURCE_ATTRIBUTES['snow.executable.name'] = '<app_name>'
  AND RECORD_TYPE = 'LOG'
  AND TIMESTAMP > DATEADD(hour, -1, CURRENT_TIMESTAMP())
ORDER BY TIMESTAMP DESC
LIMIT 100;

<event_table>``をSHOW PARAMETERSコマンドで返されたイベントテーブル名に置き換え、``<database_name>、<schema_name>、``<app_name>``をStreamlitアプリの値に置き換えます。

イベントテーブル列の詳細については、:doc:`/developer-guide/logging-tracing/event-table-columns`をご参照ください。

例:ウェアハウスランタイムアプリからのログ記録¶

import logging
import streamlit as st

logger = logging.getLogger("simple_logger")

# Write directly to the app
st.title("Simple Logging Example")

# Get the current credentials
session = st.connection('snowflake').session()

def get_log_messages_query() -> str:
    return """
            SELECT
                TIMESTAMP,
                RECORD:"severity_text"::VARCHAR AS SEVERITY,
                RESOURCE_ATTRIBUTES:"db.user"::VARCHAR AS USER,
                VALUE::VARCHAR AS VALUE
            FROM
                SAMPLE_EVENTS
            WHERE
                SCOPE:"name" = 'simple_logger'
            ORDER BY
                TIMESTAMP DESC;
            """

button = st.button("Log a message")

if button:
    try:
        logger.info("Logging an info message through Streamlit App.")
        st.success('Logged a message')
    except Exception as e:
        logger.error("Logging an error message through Streamlit App: %s",e)
        st.error('Logged an error')

sql = get_log_messages_query()

df = session.sql(sql).to_pandas()

with st.expander("**Show All Messages**"):
     st.dataframe(df, use_container_width=True)