Registro e rastreamento para Streamlit in Snowflake¶

É possível capturar mensagens de log e rastrear eventos do código do seu aplicativo Streamlit enquanto ele é executado e, em seguida, analisar os resultados com SQL, por exemplo, para analisar erros. Para obter mais informações, consulte Registro, rastreamento e métricas.

Configuração de uma tabela de eventos¶

Para registrar em log e rastrear o código do aplicativo Streamlit, é necessário criar uma tabela de evento, associar a tabela de evento a uma conta e definir os níveis de log e rastreamento de eventos. Por exemplo:

-- Create an event table if it doesn't already exist
CREATE EVENT TABLE SAMPLEDATABASE.LOGGING_AND_TRACING.SAMPLE_EVENTS;

-- Associate the event table with the account
ALTER ACCOUNT SET EVENT_TABLE = SAMPLEDATABASE.LOGGING_AND_TRACING.SAMPLE_EVENTS;

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

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

Copy

Para obter mais informações, consulte Visão geral da tabela de evento.

Exemplo: Registro de um aplicativo Streamlit¶

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:
    """
    Get data from the `EVENT TABLE` where the logs were created by this app.
    """
    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 Stremlit App.")
        st.success('Logged a message')
    except Exception as e:
        logger.error("Logging an error message through Stremlit 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)

Copy

Exemplo: Emissão de rastros de um aplicativo Streamlit¶

Nota

O exemplo a seguir requer a instalação do pacote snowflake-telemetry-python. Para obter mais informações, consulte Adição de suporte ao pacote de telemetria.

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

def sleep_function() -> int:
    """
    Function that sleeps for a random period of time, between one and ten seconds.
    """
    random_time = random.randint(1, 10)
    time.sleep(random_time)
    return random_time

def get_trace_messages_query() -> str:
    """
    Get data from the `EVENT TABLE` where the logs were created by this app.
    """
    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 :"loggin_demo.tracing" :: VARCHAR = 'begin_span'
            ORDER BY
                TIMESTAMP DESC;
            """

def trace_message() -> None:
    """
    Add a new trace message into the event table.
    """
    execution_time = sleep_function()
    telemetry.set_span_attribute("loggin_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("Succesfully 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)

Copy