Airflow：使用EmailOperator前通过Samba下载文件的方案咨询

先明确回调函数的执行节点 #

你关心的pre_execute和on_execute_callback执行节点问题，结论是：这两个回调都会在EmailOperator被分配到的worker节点上执行。Airflow中每个任务的完整生命周期（包括前置回调、任务执行、后置回调）都在它被调度到的worker节点完成，所以用这两个回调做Samba下载，确实能把文件弄到EmailOperator所在的本地节点。

但这两个方案存在明显缺陷：

• 下载逻辑和邮件任务强耦合，后续其他任务需要类似下载操作时无法复用
• 回调逻辑藏在EmailOperator的配置里，DAG可读性差，排查问题时更麻烦

自定义复合Operator的问题 #

自定义SambaToEmailOperator能把下载+发邮件封装成一个任务，但同样有复用性短板——如果之后只需要下载文件或只需要发邮件，这个Operator就无法单独使用，而且复合逻辑会增加维护成本。

推荐方案：拆分独立任务 #

最贴合Airflow「单一职责」原则的做法是拆分任务：

1. 新增一个独立的Samba下载任务，可以用PythonOperator调用SambaHook实现下载逻辑，或者封装成通用的SambaDownloadOperator方便复用
2. 任务依赖设置为：task_query >> samba_download_task >> email_task

代码示例 #

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.samba.hooks.samba import SambaHook
from airflow.operators.email import EmailOperator
from datetime import datetime

def download_from_samba():
    samba_hook = SambaHook(samba_conn_id='your_samba_conn')
    # 从Samba网络文件夹下载到EmailOperator所在节点的本地路径
    samba_hook.retrieve_file(
        remote_filepath='/network/path/query_result.csv',
        local_filepath='/local/path/query_result.csv'
    )

with DAG(dag_id='hive_query_email', start_date=datetime(2024,1,1), schedule_interval='@daily') as dag:
    task_query = # 你的Hive查询任务
    samba_download = PythonOperator(
        task_id='samba_download_file',
        python_callable=download_from_samba
    )
    email_task = EmailOperator(
        task_id='send_result_email',
        to='recipient@example.com',
        subject='Query Result',
        html_content='<p>请查看附件</p>',
        files=['/local/path/query_result.csv']
    )

    task_query >> samba_download >> email_task

如果下载逻辑需要多次复用，可以封装成自定义Operator：

from airflow.models.baseoperator import BaseOperator
from airflow.providers.samba.hooks.samba import SambaHook

class SambaDownloadOperator(BaseOperator):
    def __init__(self, samba_conn_id, remote_filepath, local_filepath, **kwargs):
        super().__init__(**kwargs)
        self.samba_conn_id = samba_conn_id
        self.remote_filepath = remote_filepath
        self.local_filepath = local_filepath

    def execute(self, context):
        samba_hook = SambaHook(self.samba_conn_id)
        samba_hook.retrieve_file(self.remote_filepath, self.local_filepath)

这个方案的优势：

• 每个任务职责清晰，DAG结构一目了然，排查问题时可单独测试下载任务
• 下载逻辑可独立复用，其他DAG需要从Samba下载时直接调用即可

内容的提问来源于stack exchange，提问作者Alexander Lopatin