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("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) 

Copy