Eğitim: Gmail gönderen operatörü

Bu öğreticide, Lakeflow Designer için bir python-run-function operatörü oluşturacak ve bir DataFrame'in içeriğini Gmail üzerinden CSV eki olarak göndereceksiniz. Bildirim gönderme veya dış sistemlere yazma gibi yan etkiler gerçekleştiren YAML tabanlı işleçler oluşturmayı öğrenmek için bu örneği kullanın. Daha fazla bilgi edinmek için bkz. Lakeflow Designer'da kullanıcı tanımlı işleçler.

Requirements

  • Bir Azure Databricks çalışma alanı ve gizli kapsamlar oluşturma erişimi.
  • Google Uygulama Parolası olan bir Gmail hesabı (çok faktörlü kimlik doğrulaması (MFA) etkinleştirildiğinde gereklidir).
  • Yerel geliştirme makinenize Databricks CLI yüklendi.

Adım 1: Gizli bilgileri ayarlayın

Gmail kimlik bilgilerinizi operatörün çalışma zamanında alabilmesi için Azure Databricks gizli dizi kapsamında depolayın.

  1. Azure Databricks CLI'ını kullanarak bir gizli anahtar kapsamı oluşturun:

    databricks secrets create-scope my_email_scope

  2. Gmail Uygulama Parolanızı şu kapsam içinde saklayın:

    databricks secrets put-secret my_email_scope gmail_app_password

    Gizli değeri girmeniz istenir. Gmail Uygulama Parolanızı yapıştırın ve kaydedin.

2. Adım: İşlevi run() yazma

İşleç python-run-function türü şu imzaya sahip bir run() işlev gerektirir:

def run(config: Dict[str, Any], inputs: Dict[str, Any], spark) -> Dict[str, Any]:
  • config: Lakeflow Designer kullanıcı arabiriminde kullanıcı tarafından sağlanan yapılandırma değerleri.
  • inputs: Bağlantı noktası adına göre anahtarlanmış giriş DataFrame'leri.
  • spark: Etkin Spark oturumu.

İşlev, çıkış bağlantı noktası adıyla anahtarlanan çıktı DataFrames sözlüğü döndürmelidir.

Bir not defteri hücresinde işlevi tanımlayın ve test edin:

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}

3. Adım: İşlevi test edin

İşlevi örnek bir DataFrame ile test edin:

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

secret_scope Yapılandırmadaki ve secret_key değerleri, gerçek parola değil, 1. Adımda oluşturduğunuz gizli dizi kapsamının ve anahtarın adlarıdır. Operatör, çalışma zamanında parolayı Azure Databricks gizli anahtarlarından almak için bu adları kullanır.

Important

Herhangi bir e-posta göndermeden geçiş davranışını doğrulamak için önce is_preview olarak ayarlanmış True ile test edin. Gerçek e-postayı test etmeye hazır olduğunuzda, is_preview değerini False olarak ayarlayın.

4. Adım: YAML tanımını oluşturma

Aşağıdaki içeriğe sahip adlı gmail_email_sender.yaml bir dosya oluşturun:

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}

5. Adım: Operatörü kaydedin ve kaydını oluşturun

  1. YAML dosyasını Azure Databricks çalışma alanınıza kaydedin. Örneğin:

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

  2. İşleci .user_defined_operators.yaml dosyanıza ekleyin:

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

Kayıt seçenekleri hakkında daha fazla bilgi için bkz. Operatörünüzü bulunabilir hale getirme.

Permissions

Bu işleci içeren bir iş akışı çalıştıran kullanıcıların gizli dizi kapsamına erişmesi gerekir READ veya işleç yapılandırmasında kendi gizli dizi kapsamlarını ve anahtar değerlerini sağlayabilirler. Kullanıcıların çalışma alanında YAML dosyasına okuma erişimine de sahip olması gerekir.

Gizli kapsama erişim izni vermek için:

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

Lakeflow Designer'da operatörü kullanın

Kayıttan sonra, işleç Lakeflow Designer'da veri kaynağınız için bir giriş bağlantı noktası ile gönderen e-postası, gizli bilgi kapsamı, gizli anahtar, alıcılar, konu ve ileti gövdesi için yapılandırma alanlarıyla birlikte görünür.

İş akışı çalıştırıldığında, işleç giriş DataFrame'i CSV'ye dönüştürür, bir e-postaya ekler ve her alıcıya gönderir. DataFrame, çıkış portuna değişmeden iletilir; böylece akışın devamında ek işleçler zincirleyebilirsiniz. İş akışı önizlemesi sırasında e-posta gönderilmez.

  • Last updated on