Journalisation et traçage pour Streamlit in Snowflake¶

Vous pouvez capturer les messages de journal et suivre les événements de votre code d’application Streamlit au fur et à mesure de son exécution, puis analyser les résultats avec SQL, par exemple, pour analyser les erreurs. Pour plus d’informations, voir Journalisation, traçage et métriques.

Configuration d’une table d’événements¶

Pour enregistrer et tracer le code de l’application Streamlit, vous devez créer une table d’événements, associer la table d’événements à un compte et définir les niveaux de journalisation et de suivi des événements. Par exemple :

-- 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

Pour plus d’informations, voir Aperçu de la table d’événements.

Exemple : journalisation à partir d’une application 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

Exemple : émission de traces à partir d’une application Streamlit¶

Note

L’exemple suivant nécessite l’installation du paquet snowflake-telemetry-python. Pour plus d’informations, voir Ajout d’une prise en charge du paquet de télémétrie.

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