Kepingan salju dapat mengirim email sekarang —jadi mari buat prosedur tersimpan yang mengirimkan hasil kueri setiap pagi — termasuk jika ada Hari Libur di suatu tempat di dunia yang memengaruhi nomor Anda.

Kepingan salju dapat mengirim email sekarang

Untuk mengirim notifikasi email dengan Snowflake, Anda sekarang hanya memerlukan satu panggilan SQL:

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

Tujuan dari posting ini adalah untuk mengikat fungsionalitas ini dengan penjadwal dan beberapa kode pemformatan — jadi setiap pagi dari Senin hingga Jumat kita bisa mendapatkan hasil kueri SQL, termasuk Hari Libur yang akan datang. Seperti yang ini:

Langkah 1: Kembangkan prosedur tersimpan yang memformat dan mengirimkan hasil kueri apa pun melalui email

Ini adalah kode yang saya tulis, sambil men-debug kemampuan saya untuk mengirim email hasil kueri yang diformat dengan baik:

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

• Snowflake sekarang memiliki “ Prosedur Anonim ” yang memungkinkan Anda mendefinisikan dan memanggilnya dalam satu langkah. Anda tidak perlu create or replacedulu dan callnanti — yang sangat berguna saat melakukan debug.
• Saya memilih untuk menulis prosedur ini dengan Python, untuk menggunakan library tabulate. Pustaka ini akan memformat hasil kueri, sehingga terlihat cantik di email keluar kita.
• tabulatePustaka sudah disediakan oleh Anaconda di Snowflake, jadi kita bisa menggunakannya dengan baris packages = (‘tabulate’).
• Dengan result_scan(last_query_id(-1)Snowflake dapat mengakses hasil dari kueri terakhir yang dieksekusi — yang memungkinkan pretty_email_results()untuk mengirim email hasil apa pun yang kami dapatkan di langkah sebelumnya.
• Snowpark memberi kita sessionobjek dalam prosedur tersimpan kita, yang dapat kita gunakan untuk menjalankan pemindaian hasil dengan session.sql().
• session.sql()memberi kami Snowpark DataFrame, dan memanggilnya .to_pandas()akan memberi kami Pandas DataFrame. Panda .to_markdown()kemudian menelepon tabulate— dan itu akan mengeluh kecuali Anda tidak secara eksplisit meminta paket ini dengan packages=(...).

Setelah proses debug interaktif selesai, Anda dapat memindahkan kode ke prosedur tersimpan permanen:

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 callerdiperlukan di sini agar prosedur tersimpan dapat menemukan riwayat kueri pengguna yang memanggilnya.
• runtime_version=3.8harus eksplisit dalam prosedur tersimpan permanen.
• Saya menambahkan kode untuk menangkap pengecualian dan mengirim email yang cantik.
• Saya memindahkan subjek dan penerima ke argumen prosedur.
• Saya menambahkan a limit 100untuk mencegah pengiriman email berukuran TB.

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

Sekarang prosedur tersimpan SQL sederhana dapat menjalankan kueri dan meminta prosedur Python kami untuk mengirimkan hasilnya melalui email:

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

• Untuk mendapatkan tabel dengan semua Hari Libur selama 60 hari ke depan , ini memanggil UDTF yang saya tentukan sebelumnya . holidays
Menghasilkan semua Hari Libur dalam SQL — dengan Python UDTF

call run_query_and_email_it();

Snowflake memiliki penjadwal tugas yang siap untuk permintaan Anda:

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

Beberapa catatan:

• Saya memintanya untuk menggunakan gudang saya yang disebut "s". Snowflake juga menawarkan Serverless Tasks , tetapi itu tidak menjalankan prosedur tersimpan Python.
• Saya harus memodifikasi holidays()UDF saya dengan memindahkan .zipdependensi Python dari tahap pengguna ke tahap bernama . Ketika itu tinggal di dalam tahap pengguna saya, tugas itu menimbulkan kesalahan Remote file ‘holidays.zip’ was not found.
• Ubah jadwal menjadi setiap 1 minutejika Anda perlu melakukan debug.

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

      
                

                    
                    
                
            

      
                

                    
                    
                
            

Jangan terlalu khawatir tentang membangun sistem peringatan — Snowflake sudah mengembangkan ini ( saat ini dalam pratinjau pribadi ):

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

• Coba ini dengan akun uji coba gratis Snowflake — Anda hanya memerlukan alamat email untuk memulai.
• Mainkan dengan opsi pemformatan yang tabulateditawarkan.