Samouczek: operator nadawcy wiadomości e-mail w usłudze Gmail

W tym samouczku tworzysz python-run-function operator dla Lakeflow Designer, który wysyła zawartość DataFrame'a jako załącznik CSV przez Gmail. Skorzystaj z tego przykładu, aby dowiedzieć się, jak tworzyć operatory oparte na języku YAML, które wykonują skutki uboczne, takie jak wysyłanie powiadomień lub zapisywanie w systemach zewnętrznych. Aby dowiedzieć się więcej, zobacz niestandardowe operatory w Lakeflow Designer.

Overview

Operator odczytuje hasło do aplikacji Gmail z zakresu wpisów tajnych Azure Databricks. Przekazuje wejściowy DataFrame bez zmian do portu wyjściowego, więc możesz dodawać kolejne operatory dalej. Podczas podglądu workflow nie wysyła żadnego maila.

Requirements

• Obszar roboczy Azure Databricks z uprawnieniami do tworzenia zakresów wpisów tajnych.
• Konto Gmail z hasłem aplikacji Google (wymagane w przypadku włączenia uwierzytelniania wieloskładnikowego).
• Interfejs wiersza polecenia (CLI) Databricks zainstalowany na lokalnej maszynie deweloperskiej.

Krok 1. Konfigurowanie wpisów tajnych

Zapisz poświadczenia Gmaila w zakresie wpisów tajnych Azure Databricks, aby operator mógł je pobrać w czasie wykonywania.

1. Utwórz zakres wpisów tajnych za pomocą interfejsu CLI usługi Azure Databricks:

   databricks secrets create-scope my_email_scope
   

2. Zapisz hasło aplikacji Gmail w obrębie zakresu:

   databricks secrets put-secret my_email_scope gmail_app_password
   

   Pojawi się monit o wprowadzenie tajnej wartości. Wklej hasło aplikacji Gmail i zapisz.

Krok 2. Pisanie run() funkcji

python-run-function Typ operatora wymaga funkcji z tym podpisemrun():

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

• config: wartości konfiguracji udostępniane przez użytkownika w interfejsie użytkownika projektanta lakeflow.
• inputs: Wejściowe ramki danych kluczowane według nazwy portu.
• spark: aktywna sesja platformy Spark.

Funkcja musi zwrócić słownik wyjściowych obiektów DataFrame, gdzie kluczami są nazwy portów wyjściowych.

Zdefiniuj i przetestuj funkcję w komórce notatnika. Gdy is_preview jest ustawiony, funkcja pomija swoje skutki uboczne i zwraca dane wejściowe bez zmian:

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}

Krok 3. Testowanie funkcji

Przetestuj funkcję na przykładowym DataFrame w trybie podglądu, który zwraca wejściowy DataFrame bez zmian i nie wysyła e-maila. Zamień <your-email@gmail.com> na swój adres Gmail.

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

result = run(
    config={
        "is_preview": True,
        "sender_email": "<your-email@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()

Krok 4. Tworzenie definicji YAML

Utwórz plik o nazwie gmail_email_sender.yaml z następującą zawartością:

schema: custom-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}

Krok 5. Zapisywanie i rejestrowanie operatora

1. Zapisz plik YAML w obszarze roboczym Azure Databricks. Przykład:

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

2. Dodaj operator do pliku .custom_operators.yaml:

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

Aby uzyskać więcej informacji o opcjach rejestracji, zobacz Zarejestruj i wdróż operator do Lakeflow Designer.

Krok 6. Konfigurowanie uprawnień

Użytkownicy, którzy uruchamiają przepływ pracy zawierający ten operator, muszą mieć READ dostęp do zakresu wpisów tajnych lub mogą podać własny zakres wpisów tajnych i wartości kluczy w konfiguracji operatora. Użytkownicy potrzebują również dostępu do odczytu do pliku YAML w obszarze roboczym.

Aby udzielić dostępu do zakresu wpisu tajnego:

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

Krok 7: Użyj operatora w Lakeflow Designer

1. W wizualnym przygotowaniu danych otwórz menu Operatorzy w lewym panelu i znajdź operatora Gmail Email Sender .
2. Przeciągnij operator na obszar roboczy i podłącz do niego wejście.
3. Kliknij dwukrotnie operatora, aby otworzyć panel konfiguracyjny, a następnie wypełnij jego pola.
4. Wybierz operatora, aby podglądać jego wyniki w panelu wyjściowym.

Operator ma port wejściowy dla źródła danych oraz pola konfiguracyjne dla e-maila nadawcy, zakresu wpisu tajnego, tajnego klucza, odbiorców, tematu i treści.

Dodatkowe zasoby

• Operatorzy niestandardowi w Lakeflow Designer
• Pisanie logiki dla niestandardowych operatorów Lakeflow Designer
• Niestandardowa referencja operatora YAML