使用Airflow EmailOperator发送S3文件时遇文件不存在错误求助

首先明确：Airflow EmailOperator本身不支持直接引用S3路径作为附件，它的files参数仅识别本地文件系统路径，这就是你遇到No such file or directory错误的原因。

以下是两种无需本地存储即可发送S3中CSV文件的方案，适配MWAA环境的资源限制：

方案1：PythonOperator + S3Hook 内存级读取发送 #

通过S3Hook直接将S3文件内容读取到内存，再调用Airflow内置的send_email函数发送邮件，全程不落地本地存储。

代码示例 #

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.amazon.aws.hooks.s3 import S3Hook
from airflow.utils.email import send_email
from datetime import datetime

def send_s3_csv_as_attachments():
    # 配置参数
    S3_BUCKET = "your-target-bucket"
    S3_CSV_KEYS = ["reports/file1.csv", "data/file2.csv"]
    RECIPIENTS = ["user@example.com", "team@example.com"]
    EMAIL_SUBJECT = "S3 CSV Files: Weekly Reports"
    EMAIL_BODY = "<p>Please find the attached CSV files from S3 storage.</p>"

    # 初始化S3Hook（需提前配置AWS连接）
    s3_hook = S3Hook(aws_conn_id="aws_default")
    
    # 构造内存附件列表
    attachments = []
    for key in S3_CSV_KEYS:
        # 读取S3文件到内存（local_path=None返回字节内容）
        file_content = s3_hook.download_file(key=key, bucket_name=S3_BUCKET, local_path=None)
        # 提取文件名
        file_name = key.split("/")[-1]
        # 添加到附件：(文件名, 字节内容, MIME类型)
        attachments.append((file_name, file_content, "text/csv"))

    # 发送邮件
    send_email(
        to=RECIPIENTS,
        subject=EMAIL_SUBJECT,
        html_content=EMAIL_BODY,
        attachments=attachments
    )

with DAG(
    dag_id="send_s3_csv_email",
    start_date=datetime(2024, 1, 1),
    schedule_interval="@weekly",
    catchup=False
) as dag:
    send_email_task = PythonOperator(
        task_id="send_s3_attachments",
        python_callable=send_s3_csv_as_attachments
    )

关键注意点 #

• 确保MWAA执行角色拥有目标S3桶的s3:GetObject权限；
• S3Hook.download_file的local_path设为None时，直接返回文件字节流，避免本地存储占用；
• 无需担心工作节点更换问题，所有操作在当前任务实例的内存中完成。

方案2：AWS SES API 直接发送（进阶） #

如果你的环境已使用AWS SES，可以调用SES的SendRawEmail API，直接引用S3对象作为附件。这种方式需要给SES配置S3访问权限，适合已有成熟SES邮件流程的场景。

核心逻辑是构造符合MIME标准的原始邮件内容，通过SES API指定S3对象作为附件来源，无需下载文件。但相比方案1，代码复杂度更高，需额外配置SES与S3的权限关联（如桶策略允许SES读取对象）。

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