在這個教學中，你會為 Lakeflow Designer 建立一個 python-run-function 操作符，透過 Gmail 將 DataFrame 的內容以 CSV 附件形式傳送。 使用此範例學習如何建立基於 YAML 的運算子，執行副作用，例如發送通知或寫入給外部系統。 欲了解更多，請參閱 Lakeflow Designer 中的使用者定義運算子。
Requirements
- 具備建立祕密範圍存取權的 Azure Databricks 工作區。
- 一個帶有 Google 應用程式密碼 的 Gmail 帳號（啟用多重驗證（MFA）時必須）。
- Databricks CLI 已安裝在您的本機開發環境中。
步驟一：設定秘密
將你的 Gmail 憑證儲存在 Azure Databricks 的秘密範圍中，讓操作員在執行時能取得。
使用 Azure Databricks CLI 建立秘密範圍：
databricks secrets create-scope my_email_scope將您的 Gmail 應用程式密碼儲存在以下範圍內：
databricks secrets put-secret my_email_scope gmail_app_password系統會提示你輸入秘密值。 貼上你的 Gmail 應用程式密碼並儲存。
步驟 2：撰寫函 run() 式
python-run-function 運算子類型需要具有以下簽名的 run() 函式：
def run(config: Dict[str, Any], inputs: Dict[str, Any], spark) -> Dict[str, Any]:
-
config： 由使用者在 Lakeflow Designer UI 中提供的設定值。 -
inputs： 輸入資料幀以埠名鍵入。 -
spark：目前作用中的 Spark 工作階段。
函式必須回傳一個以輸出埠名稱為鍵的輸出資料幀字典。
在筆記本儲存格中定義並測試這個函式：
from typing import Dict, Any
def run(config: Dict[str, Any], inputs: Dict[str, Any], spark) -> Dict[str, Any]:
input_df = inputs["data"]
# Skip side effects during Designer preview
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.")
# Retrieve password from Databricks secrets
from pyspark.dbutils import DBUtils
dbutils = DBUtils(spark)
sender_password = dbutils.secrets.get(scope=secret_scope, key=secret_key)
# Convert DataFrame to CSV
pdf = input_df.toPandas()
file_path = "/tmp/designer_email_attachment.csv"
pdf.to_csv(file_path, index=False)
# Send email to each recipient
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)
# Clean up temp file
if os.path.exists(file_path):
os.remove(file_path)
return {"data": input_df}
步驟三：測試功能
用範例資料框測試函數：
test_df = spark.createDataFrame(
[("Alice", 100), ("Bob", 200)],
["name", "amount"]
)
# Test in preview mode (no email sent)
result = run(
config={
"is_preview": True,
"sender_email": "you@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()
# Expected: the original DataFrame, unchanged
Note
設定裡的 secret_scope and secret_key 值是你在步驟 1 建立的秘密範圍和金鑰 名稱 ，不是實際密碼。 操作員會利用這些名稱在執行時從 Azure Databricks 秘密中取得密碼。
Important
先將 is_preview 設為 True 進行測試，以驗證直通行為，且不會傳送任何電子郵件。 當你準備測試實際郵件時，請設定 is_preview 為 False。
步驟 4：建立 YAML 定義
建立一個名為 gmail_email_sender.yaml 以下內容的檔案：
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}
步驟五：儲存並註冊操作員
將 YAML 檔案儲存到你的 Azure Databricks 工作區。 例如：
/Workspace/Users/<user-name>/gmail_email_sender.yaml將操作員加入你的
.user_defined_operators.yaml檔案：operators: - /Workspace/Users/<user-name>/gmail_email_sender.yaml
欲了解更多註冊選項，請參閱 讓您的營運商可被發現。
權限
執行包含此運算子的工作流程的使用者需要 READ 存取秘密作用域，或在運算子設定中提供自己的秘密作用域與金鑰值。 使用者也需要在工作區中讀取 YAML 檔案。
若要授予祕密範圍的存取權限：
databricks secrets put-acl my_email_scope <user-or-group> READ
使用Lakeflow Designer中的運算元。
註冊後，操作員會在 Lakeflow Designer 中顯示，並有資料來源的輸入埠，以及寄件人電子郵件、秘密範圍、秘密金鑰、收件人、主旨與正文的設定欄位。
當工作流程執行時，操作員會將輸入的資料框轉換成 CSV，附加到電子郵件，並寄送給每位收件人。 DataFrame 會原封不動地傳遞到輸出埠，因此你可以在下游串接更多運算子。 在工作流程預覽期間，沒有發送任何電子郵件。