O Snowflake pode enviar e-mails agora — então vamos criar um procedimento armazenado que envie a você os resultados de uma consulta todas as manhãs — inclusive se houver um feriado em algum lugar do mundo que esteja afetando seus números.

Snowflake pode enviar e-mails agora

Para enviar uma notificação por e-mail com o Snowflake, agora você precisa apenas de uma chamada SQL:

call system$send_email(
    'my_email_int',
    '[email protected], [email protected], [email protected]',
    'This is the subject',
    'This is the body'
);

O objetivo deste post é vincular essa funcionalidade a um agendador e algum código de formatação — para que todas as manhãs, de segunda a sexta-feira, possamos obter os resultados de uma consulta SQL, incluindo os próximos feriados. Como este:

Etapa 1: Desenvolva um procedimento armazenado que formate e envie por e-mail o resultado de qualquer consulta

Este é o código que escrevi, enquanto depurava minhas habilidades para enviar por e-mail um resultado de consulta bem formatado:

with pretty_email_results as procedure()
returns string
language python
packages = ('snowflake-snowpark-python', 'tabulate')
handler = 'x'
as
$$
def x(session):
    printed = session.sql(
        "select * from table(result_scan(last_query_id(-1)))"
      ).to_pandas().to_markdown()
    session.call('system$send_email',
        'my_email_int',
        '[email protected]',
        'Email Alert: Task A has finished.',
        printed)
$$
call pretty_email_results();

• O Snowflake agora tem “ procedimentos anônimos ” que permitem definir e chamá-los em uma única etapa. Você não precisa create or replaceprimeiro e calldepois — o que é especialmente útil durante a depuração.
• Eu escolhi escrever este procedimento em Python, para usar a biblioteca tabulate. Essa biblioteca formatará os resultados de uma consulta, para que fiquem bonitos em nosso e-mail de saída.
• A tabulatebiblioteca já é fornecida pelo Anaconda no Snowflake, então podemos usá-la com a linha packages = (‘tabulate’).
• Com result_scan(last_query_id(-1)o Snowflake, podemos acessar os resultados da última consulta executada - o que permite pretty_email_results()enviar por e-mail quaisquer resultados que obtivemos na etapa anterior.
• O Snowpark nos fornece o sessionobjeto em nosso procedimento armazenado, que podemos usar para executar a verificação do resultado com session.sql().
• session.sql()nos dá um Snowpark DataFrame, e invocá .to_pandas()-lo nos dá o Pandas DataFrame. O Pandas ' .to_markdown()então liga tabulate- e reclamará, a menos que você não solicite explicitamente este pacote com packages=(...).

Depois que a depuração interativa estiver concluída, você poderá mover o código para um procedimento armazenado permanente:

create or replace procedure email_last_results(send_to string, subject string)
returns string
language python
runtime_version=3.8
packages = ('snowflake-snowpark-python', 'tabulate')
handler = 'x'
execute as caller
as
$$
import snowflake

def x(session, send_to, subject):
    try: 
        body = session.sql(
          "select * from table(result_scan(last_query_id(-1))) limit 100"
        ).to_pandas().to_markdown()
    except snowflake.snowpark.exceptions.SnowparkSQLException as e:
        body = '%s\n%s' % (type(e), e)
    session.call('system$send_email',
        'my_email_int',
        send_to,
        subject,
        body)
    return 'email sent:\n%s' % body
$$;

• execute as calleré necessário aqui para que o procedimento armazenado possa localizar o histórico de consultas do usuário que o está chamando.
• runtime_version=3.8precisa ser explícito em um procedimento armazenado permanente.
• Eu adicionei código para capturar exceções e enviar um e-mail bonito de qualquer maneira.
• Mudei o assunto e os destinatários para os argumentos do procedimento.
• Eu adicionei um limit 100para evitar o envio de e-mails do tamanho de TB.

call email_last_results('[email protected]', 'results from snowflake');

Agora, um procedimento armazenado SQL simples pode executar uma consulta e solicitar que nosso procedimento Python envie os resultados por e-mail:

create or replace procedure run_query_and_email_it()
returns string
execute as caller
as
$$
begin
    -- any query you'd like to run
    select 'now' as what, to_char(current_timestamp) as value
    union all
    select 'upcoming', day || ' ' ||holiday
    from table(fh_db.public.holidays('US', [year(current_date), year(current_date)+1]))
    where day between current_date() and current_date()+60
    ;
    -- call the stored procedure that formats and emails the results
    call email_last_results('[email protected]', 'upcoming holidays');
    return 'done';
end
$$
;

• Para obter uma tabela com todos os feriados dos próximos 60 dias , ele chama meu UDTF definido anteriormente . holidays
Gerando todos os feriados em SQL — com um Python UDTF

call run_query_and_email_it();

O Snowflake tem um agendador de tarefas pronto para suas solicitações:

create or replace task email_me_frequently
warehouse = 's'
schedule = 'using cron 0 8 * * 1-5 America/Los_Angeles'
as call run_query_and_email_it();

Algumas notas:

• Pedi-lhe para usar o meu armazém chamado “s”. O Snowflake também oferece tarefas sem servidor , mas elas não executam procedimentos armazenados do Python.
• Eu tive que modificar meu holidays()UDF movendo as .zipdependências do Python de um estágio de usuário para um estágio nomeado . Quando residia dentro do meu estágio de usuário, a tarefa gerava o erro Remote file ‘holidays.zip’ was not found.
• Altere o agendamento para cada 1 minutese precisar depurar.

alter task email_me_frequently resume;

select *
  from table(information_schema.task_history(
    scheduled_time_range_start=>dateadd('hour',-48,current_timestamp()),
    result_limit => 100));

      
                

                    
                    
                
            

      
                

                    
                    
                
            

Não se preocupe muito em construir um sistema de alerta — Snowflake já está desenvolvendo isso ( atualmente em visualização privada ):

CREATE ALERT Warehouse_Credit_Usage_Alert
 WAREHOUSE = my_warehouse
 SCHEDULE = ‘USING CRON 0 7 * * *UTC’ // everyday at 7 am
 IF (EXISTS (SELECT
  Warehouse_name,
  SUM(CREDITS_USED) AS credits
  FROM snowflake.account_usage.warehouse_metering_history
  // aggregate warehouse Credit_used for the past 24 hours
 WHERE datediff(hour, start_time, CURRENT_TIMESTAMP ())<=24
 GROUP BY 1
 HAVING credits > 10
 ORDER BY 2 DESC))
 THEN system$send_email (
  ‘My_email_notification_integration’, 
        ‘admin1@company,com, [email protected]’,
        ‘Email Alert: Excessive warehouse usage!’, 
        ‘Warehouse usage exceeds 10 credits in the past 24 hours’
)

• Experimente com uma conta de avaliação gratuita do Snowflake — você só precisa de um endereço de e-mail para começar.
• Brinque com as opções de formatação que tabulateoferece.