Tutorial: Operator pengirim email Gmail

Dalam tutorial ini, Anda membuat python-run-function operator untuk Lakeflow Designer yang mengirim konten DataFrame sebagai lampiran CSV melalui Gmail. Gunakan contoh ini untuk mempelajari cara membangun operator berbasis YAML yang melakukan efek samping, seperti mengirim pemberitahuan atau menulis ke sistem eksternal. Untuk mempelajari selengkapnya, lihat Operator yang ditentukan pengguna di Lakeflow Designer.

Requirements

  • Sebuah ruang kerja Azure Databricks dengan akses untuk membuat cakupan rahasia.
  • Akun Gmail dengan Kata Sandi Google App (diperlukan saat autentikasi multifaktor (MFA) diaktifkan).
  • Databricks CLI diinstal pada komputer pengembangan lokal Anda.

Langkah 1: Menyiapkan rahasia

Simpan kredensial Gmail Anda dalam cakupan rahasia Azure Databricks sehingga operator dapat mengambilnya saat runtime.

  1. Buat cakupan rahasia menggunakan CLI Azure Databricks:

    databricks secrets create-scope my_email_scope

  2. Simpan Kata Sandi Aplikasi Gmail Anda dalam cakupan:

    databricks secrets put-secret my_email_scope gmail_app_password

    Anda diminta untuk memasukkan nilai rahasia. Tempelkan Kata Sandi Aplikasi Gmail Anda dan simpan.

Langkah 2: Tulis run() fungsi

Jenis operator python-run-function memerlukan fungsi run() dengan signatur berikut:

def run(config: Dict[str, Any], inputs: Dict[str, Any], spark) -> Dict[str, Any]:
  • config: Nilai konfigurasi yang disediakan oleh pengguna di UI Lakeflow Designer.
  • inputs: DataFrame input dengan kunci berupa nama port.
  • spark: Sesi Spark aktif.

Fungsi harus mengembalikan kamus DataFrame output dengan kunci berupa nama port output.

Tentukan dan uji fungsi dalam sel buku catatan:

from typing import Dict, Any

def run(config: Dict[str, Any], inputs: Dict[str, Any], spark) -> Dict[str, Any]:
    input_df = inputs["data"]

    # Skip side effects during Designer preview
    if config.get("is_preview", False):
        return {"data": input_df}

    import smtplib
    import os
    from email.mime.multipart import MIMEMultipart
    from email.mime.text import MIMEText
    from email.mime.base import MIMEBase
    from email import encoders

    sender_email = config.get("sender_email", "")
    secret_scope = config.get("secret_scope", "")
    secret_key = config.get("secret_key", "")
    recipients_raw = config.get("recipients", "")
    subject = config.get("subject", "")
    body = config.get("body", "")

    if not sender_email:
        raise ValueError("Sender Email is required.")
    if not secret_scope or not secret_key:
        raise ValueError("Secret Scope and Secret Key are required.")
    if not recipients_raw:
        raise ValueError("At least one recipient is required.")

    recipients = [r.strip() for r in recipients_raw.split(",") if r.strip()]
    if not recipients:
        raise ValueError("At least one valid recipient email is required.")

    # Retrieve password from Databricks secrets
    from pyspark.dbutils import DBUtils
    dbutils = DBUtils(spark)
    sender_password = dbutils.secrets.get(scope=secret_scope, key=secret_key)

    # Convert DataFrame to CSV
    pdf = input_df.toPandas()
    file_path = "/tmp/designer_email_attachment.csv"
    pdf.to_csv(file_path, index=False)

    # Send email to each recipient
    for recipient in recipients:
        msg = MIMEMultipart()
        msg["From"] = sender_email
        msg["To"] = recipient
        msg["Subject"] = subject
        msg.attach(MIMEText(body, "plain"))

        with open(file_path, "rb") as attachment:
            part = MIMEBase("application", "octet-stream")
            part.set_payload(attachment.read())
            encoders.encode_base64(part)
            part.add_header(
                "Content-Disposition",
                f"attachment; filename={os.path.basename(file_path)}",
            )
            msg.attach(part)

        with smtplib.SMTP_SSL("smtp.gmail.com", 465) as server:
            server.login(sender_email, sender_password)
            server.send_message(msg)

    # Clean up temp file
    if os.path.exists(file_path):
        os.remove(file_path)

    return {"data": input_df}

Langkah 3: Uji fungsi

Uji fungsi dengan sampel DataFrame:

test_df = spark.createDataFrame(
    [("Alice", 100), ("Bob", 200)],
    ["name", "amount"]
)

# Test in preview mode (no email sent)
result = run(
    config={
        "is_preview": True,
        "sender_email": "you@gmail.com",
        "secret_scope": "my_email_scope",
        "secret_key": "gmail_app_password",
        "recipients": "alice@example.com",
        "subject": "Test",
        "body": "Test body"
    },
    inputs={"data": test_df},
    spark=spark
)

result["data"].show()
# Expected: the original DataFrame, unchanged

Note

Nilai secret_scope dan secret_key dalam konfigurasi adalah nama cakupan rahasia dan kunci yang Anda buat di Langkah 1 -- bukan kata sandi aktual. Operator menggunakan nama-nama ini untuk mengambil kata sandi dari rahasia Azure Databricks saat runtime.

Important

Uji dengan is_preview diatur ke True terlebih dahulu untuk memverifikasi perilaku pass-through tanpa mengirim email apa pun. Saat Anda siap untuk menguji email aktual, atur is_preview ke False.

Langkah 4: Buat definisi YAML

Buat file yang disebut gmail_email_sender.yaml dengan konten berikut:

schema: user-defined-operator-v0.1.0
id: gmail_email_sender
type: python-run-function
version: '1.0.0'
name: Gmail Email Sender
description: Sends the input DataFrame as a CSV attachment via Gmail SMTP to one or more recipients.

config:
  type: object
  properties:
    is_preview:
      type: boolean
      format: is_preview
      default: false
    sender_email:
      type: string
      title: Sender Email
      default: ''
      examples:
        - 'you@gmail.com'
      x-ui:
        widget: input
    secret_scope:
      type: string
      title: Secret Scope
      default: ''
      examples:
        - 'my_email_scope'
      x-ui:
        widget: input
    secret_key:
      type: string
      title: Secret Key
      default: ''
      examples:
        - 'gmail_app_password'
      x-ui:
        widget: input
    recipients:
      type: string
      title: Recipients
      default: ''
      examples:
        - 'alice@example.com, bob@example.com'
      x-ui:
        widget: textarea
        rows: 2
    subject:
      type: string
      title: Subject
      default: ''
      examples:
        - 'Designer Output Data'
      x-ui:
        widget: input
    body:
      type: string
      title: Email Body
      default: "Hello,\n\nAttached is the latest data.\n\nBest,\nDatabricks Workflow"
      x-ui:
        widget: textarea
        rows: 6
  required:
    - sender_email
    - secret_scope
    - secret_key
    - recipients
    - subject
  additionalProperties: false

ports:
  input:
    - name: data
      title: Input Data
      mime: application/vnd.databricks.dataframe
  output:
    - name: data
      title: Output Data
      mime: application/vnd.databricks.dataframe

run_function:
  type: inline
  code: |
    from typing import Dict, Any

    def run(config: Dict[str, Any], inputs: Dict[str, Any], spark) -> Dict[str, Any]:
        input_df = inputs["data"]

        if config.get("is_preview", False):
            return {"data": input_df}

        import smtplib
        import os
        from email.mime.multipart import MIMEMultipart
        from email.mime.text import MIMEText
        from email.mime.base import MIMEBase
        from email import encoders

        sender_email = config.get("sender_email", "")
        secret_scope = config.get("secret_scope", "")
        secret_key = config.get("secret_key", "")
        recipients_raw = config.get("recipients", "")
        subject = config.get("subject", "")
        body = config.get("body", "")

        if not sender_email:
            raise ValueError("Sender Email is required.")
        if not secret_scope or not secret_key:
            raise ValueError("Secret Scope and Secret Key are required.")
        if not recipients_raw:
            raise ValueError("At least one recipient is required.")

        recipients = [r.strip() for r in recipients_raw.split(",") if r.strip()]
        if not recipients:
            raise ValueError("At least one valid recipient email is required.")

        from pyspark.dbutils import DBUtils
        dbutils = DBUtils(spark)
        sender_password = dbutils.secrets.get(scope=secret_scope, key=secret_key)

        pdf = input_df.toPandas()
        file_path = "/tmp/designer_email_attachment.csv"
        pdf.to_csv(file_path, index=False)

        for recipient in recipients:
            msg = MIMEMultipart()
            msg["From"] = sender_email
            msg["To"] = recipient
            msg["Subject"] = subject
            msg.attach(MIMEText(body, "plain"))

            with open(file_path, "rb") as attachment:
                part = MIMEBase("application", "octet-stream")
                part.set_payload(attachment.read())
                encoders.encode_base64(part)
                part.add_header(
                    "Content-Disposition",
                    f"attachment; filename={os.path.basename(file_path)}",
                )
                msg.attach(part)

            with smtplib.SMTP_SSL("smtp.gmail.com", 465) as server:
                server.login(sender_email, sender_password)
                server.send_message(msg)

        if os.path.exists(file_path):
            os.remove(file_path)

        return {"data": input_df}

Langkah 5: Simpan dan daftarkan operator

  1. Simpan file YAML ke ruang kerja Azure Databricks Anda. Contohnya:

    /Workspace/Users/<user-name>/gmail_email_sender.yaml

  2. Tambahkan operator ke file .user_defined_operators.yaml Anda:

    operators:
  - /Workspace/Users/<user-name>/gmail_email_sender.yaml

Untuk informasi selengkapnya tentang opsi pendaftaran, lihat Membuat operator Anda dapat ditemukan.

Permissions

Pengguna yang menjalankan alur kerja yang berisi operator ini memerlukan READ akses ke cakupan rahasia, atau mereka dapat memberikan cakupan rahasia dan nilai kunci mereka sendiri dalam konfigurasi operator. Pengguna juga memerlukan akses baca ke file YAML di ruang kerja.

Untuk memberikan akses ke lingkup rahasia:

databricks secrets put-acl my_email_scope <user-or-group> READ

Menggunakan operator di Lakeflow Designer

Setelah pendaftaran, operator muncul di Lakeflow Designer dengan port input untuk sumber data dan bidang konfigurasi Anda untuk email pengirim, cakupan rahasia, kunci rahasia, penerima, subjek, dan isi.

Saat alur kerja berjalan, operator mengonversi input DataFrame ke CSV, melampirkannya ke email, dan mengirimkannya ke setiap penerima. DataFrame diteruskan ke port output tanpa perubahan, sehingga Anda dapat merangkai operator tambahan setelahnya. Selama pratinjau alur kerja, tidak ada email yang dikirim.

  • Last updated on