Didacticiel : opérateur d'expéditeur d'e-mail Gmail

Dans ce tutoriel, vous créez un opérateur python-run-function pour Lakeflow Designer qui envoie le contenu d'un DataFrame en tant que pièce jointe au format CSV via Gmail. Utilisez cet exemple pour apprendre à créer des opérateurs basés sur YAML qui effectuent des effets secondaires, comme l'envoi de notifications ou l'écriture sur des systèmes externes. Pour en savoir plus, consultez Opérateurs personnalisés dans Lakeflow Designer.

Présentation

L'opérateur lit le mot de passe de l'application Gmail à partir d'un Secret Scope Databricks. Il transmet le DataFrame d’entrée tel quel à son port de sortie, de sorte que vous pouvez chaîner des opérateurs supplémentaires en aval. Lors d'une prévisualisation de workflow, il n'envoie pas d'e-mail.

Exigences

• Un Databricks Workspace avec accès pour créer des Secret Scope.
• Un compte Gmail avec un mot de passe d'application Google (requis lorsque l'authentification multifacteur (MFA) est activée).
• La CLI Databricks installée sur votre machine de développement locale.

Étape 1 : Configurez les secrets

Stockez vos identifiants Gmail dans un Secret Scope Databricks afin que l'opérateur puisse les récupérer au moment de l'exécution.

1. Créer un Secret Scope à l'aide de l'interface CLI de Databricks :

   Bash

   databricks secrets create-scope my_email_scope

2. Stockez votre mot de passe d’application Gmail dans l’étendue :

   Bash

   databricks secrets put-secret my_email_scope gmail_app_password

   Vous êtes invité à saisir la valeur secrète. Collez votre mot de passe d’application Gmail et enregistrez.

Étape 2 : Écrire la fonction run()

Le type d’opérateur python-run-function nécessite une fonction run() avec cette signature :

Python

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

• config : valeurs de configuration fournies par l'utilisateur dans l'interface utilisateur de Lakeflow Designer.
• inputs : DataFrames d'entrée indexés par nom de port.
• spark : la session Spark active.

La fonction doit renvoyer un dictionnaire de DataFrames de sortie indexés par nom de port de sortie.

Définissez et testez la fonction dans une cellule de notebook. Lorsque is_preview est défini, la fonction ignore ses effets secondaires et renvoie l’entrée inchangée :

Python

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}

Étape 3 : Testez la fonction

Testez la fonction avec un DataFrame d’exemple en mode aperçu, ce qui renvoie le DataFrame d’entrée inchangé et n’envoie aucun e-mail. Remplacez <your-email@gmail.com> par votre adresse Gmail.

Python

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

remarque

Les valeurs secret_scope et secret_key dans la configuration sont les noms du Secret Scope et de la clé que vous avez créés lors de l’étape 1, et non le mot de passe réel. L’opérateur utilise ces noms pour récupérer le mot de passe à partir des secrets Databricks au moment de l’exécution.

important

Testez d'abord avec is_preview défini sur True pour vérifier le comportement de transmission sans envoyer d'e-mail. Lorsque vous êtes prêt à tester l'e-mail réel, définissez is_preview sur False.

Étape 4 : créer la définition YAML

Créez un fichier appelé gmail_email_sender.yaml avec le contenu suivant :

YAML

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}

Étape 5 : Enregistrer et enregistrer l'opérateur

1. Enregistrez le fichier YAML dans votre workspace Databricks. Par exemple :

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

2. Ajoutez l'opérateur à votre fichier .custom_operators.yaml :

   YAML

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

Pour plus d'informations sur les options d'enregistrement, consultez Enregistrer et déployer votre opérateur dans Lakeflow Designer.

Étape 6 : configurez les autorisations

Les utilisateurs qui exécutent un workflow contenant cet opérateur nécessitent un accès READ au Secret Scope, ou ils peuvent fournir leurs propres Secret Scope et valeurs de clé dans la configuration de l'opérateur. Les utilisateurs ont également besoin d’un accès en lecture au fichier YAML dans le workspace.

Pour accorder l'accès au Secret Scope :

Bash

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

Étape 7 : Utilisez l’opérateur dans Lakeflow Designer

1. Dans une préparation visuelle des données, ouvrez le menu Opérateurs dans le panneau de gauche et recherchez l’opérateur Expéditeur d’e-mails Gmail .
2. Faites glisser l'opérateur sur la toile et connectez-lui une entrée.
3. Double-cliquez sur l'opérateur pour ouvrir son volet de configuration, puis définissez ses champs.
4. Sélectionnez l’opérateur pour prévisualiser ses résultats dans le volet de sortie.

L’opérateur dispose d’un port d’entrée pour votre source de données et de champs de configuration pour l’e-mail de l’expéditeur, le Secret Scope, la clé secrète, les destinataires, l’objet et le corps du message.

Ressources supplémentaires

• Opérateurs personnalisés dans Lakeflow Designer
• Écrire la logique pour les opérateurs personnalisés de Lakeflow Designer
• Référence YAML de l’opérateur personnalisé