Tutorial: Operador de envio email do Gmail

Neste tutorial, você criará um operador python-run-function para o Lakeflow Designer que envia o conteúdo de um DataFrame como um anexo CSV através do Gmail. Use este exemplo para saber como criar operadores baseados em YAML que realizam efeitos colaterais, como enviar notificações ou gravar em sistemas externos. Para saber mais, consulte Operadores personalizados no Lakeflow Designer.

Visão geral

O operador lê a senha do aplicativo do Gmail de um Secret Scope da Databricks. Ele passa o DataFrame de entrada sem alterações para sua porta de saída, para que você possa encadear operadores adicionais downstream. Durante uma prévia de fluxo de trabalho, ele não envia email.

Requisitos

• Um workspace Databricks com acesso para criar Escopo Secreto.
• Uma account do Gmail com uma senha de aplicativo do Google (necessária quando a autenticação multifator (MFA) está ativada).
• A CLI do Databricks instalada em sua máquina de desenvolvimento local.

o passo 1: Configurar segredos

Armazene suas credenciais do Gmail em um Escopo Secreto Databricks para que o operador possa recuperá-las em tempo de execução.

1. Criar um escopo secreto usando a CLI Databricks :

   Bash

   databricks secrets create-scope my_email_scope

2. Armazene a senha do seu aplicativo Gmail no seguinte escopo:

   Bash

   databricks secrets put-secret my_email_scope gmail_app_password

   Você será solicitado a inserir o valor secreto. Cole a senha do seu aplicativo Gmail e salve.

o passo 2: Escreva a função run()

O tipo de operador python-run-function requer uma função run() com esta assinatura:

Python

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

• config : Valores de configuração fornecidos pelo usuário na interface do Lakeflow Designer.
• inputs : DataFrames de entrada indexados pelo nome da porta.
• spark : A sessão Spark ativa.

A função deve retornar um dicionário de DataFrames de saída indexados pelo nome da porta de saída.

Defina e teste a função em uma célula do notebook. Quando is_preview está definido, a função pula seus efeitos colaterais e retorna a entrada inalterada:

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}

o passo 3: Teste a função

Teste a função com um DataFrame de amostra no modo de visualização, que retorna o DataFrame de entrada inalterado e não envia nenhum email. Substitua <your-email@gmail.com> pelo seu endereço do 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()

nota

Os valores secret_scope e secret_key na configuração são os nomes do Secret Scope e da key que você criou no passo 1, e não a senha real. O operador usa esses nomes para recuperar a senha dos segredos do Databricks no Runtime.

importante

Teste primeiro com is_preview definido como True para verificar o comportamento de passagem sem enviar nenhum email. Quando você estiver pronto para testar o email real, defina is_preview para False.

o passo 4: Construir a definição YAML

Crie um arquivo chamado gmail_email_sender.yaml com o seguinte conteúdo:

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}

o passo 5: Salvar e registrar a operadora

1. Salve o arquivo YAML em seu workspace Databricks . Por exemplo:

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

2. Adicione o operador ao seu arquivo .custom_operators.yaml :

   YAML

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

Para obter mais informações sobre as opções de registro, consulte Registrar e implantar seu operador no Lakeflow Designer.

O passo 6: configurar permissões

Usuários que executam um fluxo de trabalho contendo este operador precisam de acesso READ ao Escopo Secreto, ou podem fornecer seu próprio Escopo Secreto e valores key na configuração do operador. Os usuários também precisam de acesso de leitura ao arquivo YAML no workspace.

Para conceder acesso ao Escopo Secreto:

Bash

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

o passo 7: usar o operador no Lakeflow Designer

1. Na preparação visual de dados, abra o menu Operators no painel esquerdo e encontre o operador Gmail Email Sender .
2. Arraste o operador para a tela e conecte uma entrada a ele.
3. Clique duas vezes no operador para abrir seu painel de configuração e, em seguida, defina seus campos.
4. Selecione o operador para visualizar seus resultados no painel de saída.

O operador tem uma porta de entrada para sua fonte de dados e campos de configuração para o email do remetente, Secret Scope, chave secreta, destinatários, assunto e corpo.

Recursos adicionais

• Operadores personalizados no Lakeflow Designer
• Escrever lógica para operadores personalizados do Lakeflow Designer
• Referência de YAML do operador personalizado