Airflow Kubernetes Executor下跨任务共享临时文件并邮件发送咨询

针对你使用Kubernetes Executor时，跨Pod任务共享文件给EmailOperator的需求，这里提供两个无需额外挂载持久化存储的简便方案：

方案一：用XCom传递文件内容（适合小文件） #

利用Airflow内置的XCom机制传递文件内容，无需额外依赖：

1. 调整Bash Task A：生成文件后读取内容输出到stdout，Airflow会自动将stdout内容推送到XCom（需开启do_xcom_push=True）

   task_a = BashOperator(
       task_id="generate_file",
       bash_command="""
           python3 -c "
               # 替换为你的文件生成逻辑
               with open('/tmp/output.txt', 'w') as f:
                   f.write('你的文件内容')
               # 读取内容并输出，供XCom捕获
               with open('/tmp/output.txt', 'r') as f:
                   print(f.read())
           "
       """,
       do_xcom_push=True,
       dag=dag
   )
   

2. 新增中间任务+Email Task B：先从XCom拉取内容生成临时文件，再用EmailOperator发送附件

   def create_temp_file(**context):
       # 获取Task A的XCom内容
       file_content = context['ti'].xcom_pull(task_ids='generate_file')
       temp_file_path = '/tmp/email_attachment.txt'
       with open(temp_file_path, 'w') as f:
           f.write(file_content)
       # 传递临时文件路径给下一个任务
       return temp_file_path
   
   prepare_attachment = PythonOperator(
       task_id="prepare_attachment",
       python_callable=create_temp_file,
       provide_context=True,
       dag=dag
   )
   
   task_b = EmailOperator(
       task_id="send_email",
       to="recipient@example.com",
       subject="带附件的邮件",
       html_content="<p>请查看附件</p>",
       files="{{ ti.xcom_pull(task_ids='prepare_attachment') }}",
       dag=dag
   )
   
   # 设置任务依赖
   task_a >> prepare_attachment >> task_b
   

方案二：用临时对象存储（适合大文件） #

如果文件体积超过XCom默认48KB限制，可借助轻量对象存储（如MinIO）实现跨Pod共享：

1. Task A上传文件到对象存储：在Bash命令中加入上传逻辑（假设已安装MinIO客户端）

   # 生成文件后执行上传
   mc cp /tmp/output.txt minio/temp-bucket/$(date +%Y%m%d%H%M%S)_output.txt
   # 输出文件存储路径到stdout，通过XCom传递给Task B
   echo "minio/temp-bucket/$(date +%Y%m%d%H%M%S)_output.txt"
   

2. Task B下载文件并发送：用PythonOperator先下载文件，再调用EmailOperator发送附件，完成后可删除临时文件避免存储占用。

内容的提问来源于stack exchange，提问作者Deepak Kothari