Tutorial: Operador de remitente de correo electrónico de Gmail

En este tutorial, creas un python-run-function operador para Lakeflow Designer que envía el contenido de un DataFrame como archivo adjunto CSV a través de Gmail. Use este ejemplo para aprender a crear operadores basados en YAML que realizan efectos secundarios, como enviar notificaciones o escribir en sistemas externos. Para más información, consulte Operadores definidos por el usuario en Lakeflow Designer.

Información general

El operador lee la contraseña de la app de Gmail desde un ámbito secreto de Azure Databricks. Pasa el DataFrame de entrada sin cambios a su puerto de salida, así que puedes encadenar operadores adicionales aguas abajo. Durante una vista previa del flujo de trabajo, no envía ningún correo.

Requirements

• Un área de trabajo de Azure Databricks con acceso para crear ámbitos secretos.
• Una cuenta de Gmail con una contraseña de aplicación de Google (necesaria cuando la autenticación multifactor (MFA) está habilitada).
• La CLI de Databricks instalada en su máquina de desarrollo local.

Paso 1: Configuración de secretos

Almacene sus credenciales de Gmail en un ámbito secreto de Azure Databricks para que el operador pueda recuperarlas en tiempo de ejecución.

1. Cree un ámbito de secreto mediante la CLI de Azure Databricks:

   databricks secrets create-scope my_email_scope
   

2. Guarda tu contraseña de aplicación de Gmail en el ámbito:

   databricks secrets put-secret my_email_scope gmail_app_password
   

   Se le pedirá que escriba el valor del secreto. Pegue la contraseña de la aplicación de Gmail y guarde.

Paso 2: Escribir la run() función

El tipo de python-run-function operador requiere una run() función con esta firma:

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

• config: valores de configuración proporcionados por el usuario en la interfaz de usuario del Diseñador de Lakeflow.
• inputs: Introducir DataFrames indexados por nombre de puerto.
• spark: la sesión activa de Spark.

La función debe devolver un diccionario de DataFrames de salida cuyas claves sean los nombres de los puertos de salida.

Define y prueba la función en una celda de cuaderno. Cuando is_preview está activado, la función omite sus efectos secundarios y devuelve la entrada sin cambios:

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}

Paso 3: Probar la función

Prueba la función con un DataFrame de ejemplo en modo de vista previa, que devuelve el DataFrame de entrada sin cambios y no envía ningún correo electrónico. Sustituye <your-email@gmail.com> por tu dirección de 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()

Paso 4: Compilación de la definición de YAML

Cree un archivo llamado gmail_email_sender.yaml con el siguiente contenido:

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}

Paso 5: Guardar y registrar el operador

1. Guarde el archivo YAML en el área de trabajo de Azure Databricks. Por ejemplo:

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

2. Añada el operador a su archivo .user_defined_operators.yaml:

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

Para más información sobre las opciones de registro, consulte Registrar e implementar su operador en Lakeflow Designer.

Paso 6: Configuración de permisos

Los usuarios que ejecutan un flujo de trabajo que contiene este operador necesitan READ acceso al ámbito del secreto, o pueden proporcionar su propio ámbito secreto y valores de clave en la configuración del operador. Los usuarios también necesitan acceso de lectura al archivo YAML del área de trabajo.

Para conceder acceso al ámbito secreto:

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

Paso 7: Utiliza el operador en Lakeflow Designer

1. En una preparación visual de datos, abre el menú de Operadores en el panel izquierdo y busca el operador Gmail Email Sender.
2. Arrastra al operador sobre el lienzo y conecta una entrada a él.
3. Haz doble clic en el operador para abrir su panel de configuración y luego establece sus campos.
4. Selecciona el operador para previsualizar sus resultados en el panel de salida.

El operador tiene un puerto de entrada para tu fuente de datos y campos de configuración para correo electrónico del remitente, alcance secreto, clave secreta, destinatarios, asunto y cuerpo.

Recursos adicionales

• Operadores definidos por el usuario en Lakeflow Designer
• Escribir la lógica para operadores personalizados de Lakeflow Designer
• Referencia de YAML del operador definido por el usuario