Logging and tracing for Streamlit in Snowflake¶

Streamlit in Snowflake supports logging for both warehouse and container runtimes. Warehouse runtimes use the Snowflake telemetry framework to capture log messages and trace events into an event table. Container runtimes capture logs that your app emits to standard output and standard error, store them in the account’s event table, and provide both live console logs and historical log views in Snowsight.

Both runtimes store logs in the account-level event table. An account administrator must set up and configure this event table before logs can be captured. For instructions, see Event table overview.

• 要查找为您的账户配置的事件表，请运行：
  Copy codeExpand code block

  SHOW PARAMETERS LIKE 'event_table' IN ACCOUNT;

  Show lessSee more

  Scroll to top

下表按运行时比较了日志记录和跟踪支持：

容器运行时日志记录

Container-runtime Streamlit apps run inside a Snowpark Container Services container. Snowflake automatically captures anything your app emits to standard output and standard error and stores it in the account’s event table. You can view these logs in Snowsight or query them with SQL.

Python 的日志记录模块¶

Use Python’s built-in logging module to emit log messages from your app. The following example configures a logger that writes INFO-level and higher messages to standard output:

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)

Live logs in Snowsight¶

When you edit a container-runtime app in Snowsight, a logs pane appears below the editor. This pane streams log messages in real time as your app emits them. A short history of the most recent logs is displayed when you first connect.

每个日志条目会显示以下信息：

可用的实时日志操作

在日志窗格的右上角，您可以搜索和筛选日志，以帮助您找到所需的信息。这包括文本搜索、按来源筛选以及按严重性级别筛选。您可以在三点菜单中，下载当前日志、导航到历史日志或清除实时日志记录窗格。当您清除窗格时，当前日志将从当前视图中删除，但不会从事件表中删除。立即重新加载页面将恢复最近的日志。

了解日志来源

来自容器运行时应用程序的日志有以下两个来源之一：

• MANAGER: The system process inside the container that prepares and runs your app. Manager logs include messages about downloading your app files from the stage, installing Python dependencies, and starting the Streamlit server process. If you update your app’s dependency files while the app is running, the manager process reinstalls dependencies and produces additional manager logs.
• APP: Logs from the running Streamlit server process. This includes messages from your user-configured Python loggers, Streamlit’s built-in logger, and any other output your app writes to standard output or standard error.

The boundary between sources is the streamlit run command. Everything the container does before starting the Streamlit process produces MANAGER logs. After the Streamlit process starts, output from that process produces APP logs.

View historical logs in Snowsight¶

以下步骤仅适用于容器运行时应用程序。仓库运行时应用程序没有日志窗格。

1. Sign in to Snowsight.
2. In the navigation menu, select Projects » Streamlit, and then select your app.
3. In the upper-right corner of the page, select Edit.
4. In the upper-right corner of the logs pane, select the three-dot menu (Other actions) » Historical logs.

This opens the Snowpark Container Services monitoring page for the service that runs behind your app. The logs table shows the following columns:

For more information about the monitoring page, see 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;

Replace <event_table> with the event table name returned by the SHOW PARAMETERS command, and replace <database_name>, <schema_name>, and <app_name> with the values for your Streamlit app.

For more information about the event table columns, see Event table columns.

仓库运行时日志记录

For Streamlit apps using warehouse runtimes, you can capture log messages and trace events of your Streamlit app code as it runs and then analyze the results with SQL, for example, to analyze errors. For more information, see Logging, tracing, and metrics.

仓库运行时需要在包含您应用程序的数据库上设置日志和跟踪级别：

-- Set the log level for the database containing your app
ALTER DATABASE <database_name> SET LOG_LEVEL = INFO;

-- Set the trace level for the database containing your app
ALTER DATABASE <database_name> SET TRACE_LEVEL = ON_EVENT;

示例：从仓库运行时应用程序记录日志

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)

跟踪（仅限仓库运行时）

仅仓库运行时支持跟踪。您可以从 Streamlit 应用程序发出跟踪事件，然后查询事件表以对其进行分析。

import streamlit as st
import time
import random
from snowflake import telemetry

def sleep_function() -> int:
    random_time = random.randint(1, 10)
    time.sleep(random_time)
    return random_time

def get_trace_messages_query() -> str:
    return """
            SELECT
                TIMESTAMP,
                RESOURCE_ATTRIBUTES :"db.user" :: VARCHAR AS USER,
                RECORD_TYPE,
                RECORD_ATTRIBUTES
            FROM
                SAMPLE_EVENTS
            WHERE
                RECORD :"name" :: VARCHAR = 'tracing_some_data'
                OR RECORD_ATTRIBUTES :"logging_demo.tracing" :: VARCHAR = 'begin_span'
            ORDER BY
                TIMESTAMP DESC;
            """

def trace_message() -> None:
    execution_time = sleep_function()
    telemetry.set_span_attribute("logging_demo.tracing", "begin_span")
    telemetry.add_event(
        "tracing_some_data",
        {"function_name": "sleep_function", "execution_time": execution_time},
    )

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

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

button = st.button("Add trace event")

if button:
    with st.spinner("Executing function..."):
        trace_message()
        st.toast("Successfully log a trace message!", icon="✅")

sql = get_trace_messages_query()

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

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