Airflow Email Operator附件配置问题：文件路径与语法咨询

Great question! Let's break this down clearly to fix your "file not found" error:

Where does Airflow read the files from? #

By default, the files parameter in the EmailOperator looks for files on the local filesystem of the Airflow worker node that executes your task.

If you're using Google Cloud Composer:

• Your DAG files stored in the Composer GCS bucket's dags/ directory are automatically synced to the worker's local path /home/airflow/gcs/dags/. But note: only files directly in or nested under dags/ get synced—files in other GCS bucket directories (like data/ or plugins/) won't be present on the worker's local filesystem unless you explicitly download them.

Where should you upload your file? #

You have two main options depending on your use case:

• Directly to the Composer GCS bucket's dags/ directory
  This is the simplest approach if the file is tied to your DAG (e.g., a template or small attachment). Composer will sync it to the worker's local /home/airflow/gcs/dags/ path automatically (usually within a few minutes of upload).
• To another GCS bucket/directory (e.g., data/)
  If your file lives in a different GCS location, you'll need to first download it to the worker's local filesystem using an operator like GCSToLocalFilesystemOperator before passing it to the EmailOperator.

Correct syntax for the files parameter #

Let's cover both scenarios with examples:

Scenario 1: File is in the dags/ directory (synced to worker) #

If abc.txt is in the same directory as your DAG file in GCS dags/, you can use either a relative path (relative to your DAG file) or an absolute path:

from airflow.operators.email import EmailOperator

email_task = EmailOperator(
    task_id="send_email_with_attachment",
    to="your-recipient@example.com",
    subject="Attachment Test",
    html_content="<p>See attached file!</p>",
    # Option 1: Relative path (works if file is in same directory as DAG)
    files=["abc.txt"],
    # Option 2: Absolute path to the synced directory
    # files=["/home/airflow/gcs/dags/abc.txt"]
)

Scenario 2: File is in another GCS location #

First download the file to the worker's local filesystem (we'll use /tmp/ since it's a temporary directory that gets cleaned up automatically), then attach it:

from airflow.operators.email import EmailOperator
from airflow.providers.google.cloud.transfers.gcs_to_local import GCSToLocalFilesystemOperator

# Step 1: Download file from GCS to worker local
download_file = GCSToLocalFilesystemOperator(
    task_id="download_abc_txt",
    bucket="your-composer-environment-bucket",
    object_name="data/abc.txt",  # Path to file in GCS
    filename="/tmp/abc.txt"  # Local path on worker to save the file
)

# Step 2: Send email with the downloaded file
send_email = EmailOperator(
    task_id="send_email_with_gcs_attachment",
    to="your-recipient@example.com",
    subject="GCS File Attachment",
    html_content="<p>Attached is the file from GCS!</p>",
    files=["/tmp/abc.txt"]
)

# Set task dependency: download first, then send email
download_file >> send_email

Quick Notes #

• Always verify that the file has finished syncing to the worker before running your task (for dags/ directory uploads, give it 2-3 minutes).
• For larger files, avoid storing them in the dags/ directory—use the GCS-to-local download approach instead to keep your DAG directory clean.

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