Zelfstudie: Operator voor e-mailzender van Gmail

In deze tutorial maak je een python-run-function operator aan voor Lakeflow Designer die de inhoud van een DataFrame als CSV-bijlage via Gmail stuurt. In dit voorbeeld leert u hoe u op YAML gebaseerde operators bouwt die bijwerkingen uitvoeren, zoals het verzenden van meldingen of schrijven naar externe systemen. Zie Door de gebruiker gedefinieerde operators in Lakeflow Designer voor meer informatie.

Overview

De operator leest het Gmail App Password uit een Azure Databricks secret scope. Het laat het invoerdataframe ongewijzigd doorgaan naar zijn uitvoerpoort, zodat je extra operatoren stroomafwaarts kunt koppelen. Tijdens een workflow-preview wordt er geen e-mail gestuurd.

Requirements

• Een Azure Databricks-werkruimte met toegang om secret scopes te maken.
• Een Gmail-account met een Google App-wachtwoord (vereist wanneer meervoudige verificatie (MFA) is ingeschakeld).
• De Databricks CLI geïnstalleerd op uw lokale ontwikkelcomputer.

Stap 1: Geheime gegevens instellen

Sla uw Gmail-referenties op in een Azure Databricks geheim bereik, zodat de operator deze tijdens runtime kan ophalen.

1. Maak een geheim bereik met behulp van de Azure Databricks CLI:

   databricks secrets create-scope my_email_scope
   

2. Sla uw Gmail-appwachtwoord op in de scope:

   databricks secrets put-secret my_email_scope gmail_app_password
   

   U wordt gevraagd de geheime waarde in te voeren. Plak uw Gmail-appwachtwoord en sla het op.

Stap 2: De run() functie schrijven

Voor het python-run-function operatortype is een run() functie met deze handtekening vereist:

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

• config: Configuratiewaarden die door de gebruiker worden verstrekt in de gebruikersinterface van Lakeflow Designer.
• inputs: Invoer-DataFrames met poortnaam als sleutel.
• spark: De actieve Spark-sessie.

De functie moet een woordenlijst met uitvoerdataframes retourneren die zijn gekoppeld aan de naam van de uitvoerpoort.

Definieer en test de functie in een notitieboekcel. Wanneer is_preview ingesteld is, slaat de functie zijn bijwerkingen over en geeft de invoer ongewijzigd terug:

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}

Stap 3: De functie testen

Test de functie met een voorbeeld DataFrame in preview-modus, die de invoer DataFrame ongewijzigd teruggeeft en geen e-mail stuurt. Vervang <your-email@gmail.com> het met je Gmail-adres.

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

Stap 4: de YAML-definitie bouwen

Maak een bestand gmail_email_sender.yaml met de naam met de volgende inhoud:

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}

Stap 5: de operator opslaan en registreren

1. Sla het YAML-bestand op in uw Azure Databricks werkruimte. Voorbeeld:

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

2. Voeg de operator toe aan uw .user_defined_operators.yaml bestand:

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

Voor meer informatie over registratieopties, zie Registreren en implementeer je operator in Lakeflow Designer.

Stap 6: Machtigingen instellen

Gebruikers die een werkstroom uitvoeren die deze operator bevat, moeten toegang hebben READ tot het geheime bereik of ze kunnen hun eigen geheime bereik en sleutelwaarden opgeven in de operatorconfiguratie. Gebruikers hebben ook leestoegang nodig tot het YAML-bestand in de werkruimte.

Toegang verlenen tot het geheime bereik:

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

Stap 7: Gebruik de operator in Lakeflow Designer

1. Open bij een visuele gegevensvoorbereiding het Operators-menu in het linkerpaneel en zoek de operator Gmail Email Sender .
2. Sleep de operator op het canvas en verbind een invoer ermee.
3. Dubbelklik op de operator om het configuratiepaneel te openen en stel vervolgens de velden in.
4. Selecteer de operator om de bijbehorende resultaten in het uitvoervenster te bekijken.

De operator heeft een invoerpoort voor uw databron en configuratievelden voor afzender-e-mail, geheime scope, geheime sleutel, ontvangers, onderwerp en hoofdtekst.

Aanvullende bronnen

• Door de gebruiker gedefinieerde operators in Lakeflow Designer
• Schrijf logica voor door Lakeflow Designer door gebruikers gedefinieerde operatoren
• YAML-verwijzing voor door de gebruiker gedefinieerde operators