Cloud Composer临时文件上传GCS失败，咨询替代实现方案

核心原因 #

Cloud Composer（托管式Airflow）的/tmp目录是任务执行节点的本地临时存储，会被系统定期清理；更关键的是，不同任务可能分配到不同worker节点运行——如果任务1和任务2不在同一个worker上，任务2根本找不到任务1生成的文件，这才是导致偶发文件缺失的主要原因。

推荐实现方案 #

方案1：XCom传递内容（小文件场景） #

如果CSV文件体积较小（建议<100MB），可以直接通过Airflow的XCom在任务间传递文件内容，完全绕开本地文件依赖：

• 任务1生成数据后，直接返回文件内容或序列化对象
• 任务2从XCom中读取内容，直接写入GCS

示例代码：

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.google.cloud.hooks.gcs import GCSHook
from datetime import datetime

def generate_csv_content():
    # 模拟生成CSV内容
    csv_content = "col1,col2\nval1,val2\nval3,val4"
    return csv_content

def upload_to_gcs(**context):
    csv_content = context['ti'].xcom_pull(task_ids='generate_csv')
    gcs_hook = GCSHook()
    # 直接将内容写入GCS对象
    gcs_hook.upload(
        bucket_name='your-bucket-name',
        object_name='path/to/abc.csv',
        data=csv_content.encode('utf-8')
    )

with DAG(
    dag_id='csv_to_gcs',
    start_date=datetime(2024,1,1),
    schedule_interval=None,
    catchup=False
) as dag:
    generate_task = PythonOperator(
        task_id='generate_csv',
        python_callable=generate_csv_content
    )

    upload_task = PythonOperator(
        task_id='upload_to_gcs',
        python_callable=upload_to_gcs,
        provide_context=True
    )

generate_task >> upload_task

方案2：使用共享存储路径（大文件场景） #

如果文件体积较大，XCom不适用，可改用Cloud Composer集群的共享存储目录/home/airflow/gcs/data/——这个目录挂载了Composer关联的GCS存储桶的data前缀，所有worker节点都能访问，且不会被系统自动清理。

修改步骤：

1. 任务1将文件生成到/home/airflow/gcs/data/abc.csv
2. 任务2从该路径读取文件上传GCS（甚至无需手动上传：写入该目录的文件会自动同步到GCS的gs://<your-composer-bucket>/data/路径）

示例代码（手动上传方式）：

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.providers.google.cloud.operators.gcs import GCSSynchronizeBucketsOperator
from datetime import datetime

with DAG(
    dag_id='csv_to_gcs_shared',
    start_date=datetime(2024,1,1),
    schedule_interval=None,
    catchup=False
) as dag:
    generate_csv = BashOperator(
        task_id='generate_csv',
        bash_command='echo "col1,col2\nval1,val2" > /home/airflow/gcs/data/abc.csv'
    )

    upload_to_gcs = GCSSynchronizeBucketsOperator(
        task_id='upload_to_gcs',
        source_bucket='your-composer-bucket',
        source_object='data/abc.csv',
        destination_bucket='your-target-bucket',
        destination_object='final/path/abc.csv'
    )

generate_csv >> upload_to_gcs

方案3：合并为单个任务（简单场景） #

如果两个任务逻辑不复杂，直接合并成一个Python任务：在同一个任务内完成文件生成与GCS上传，彻底避免跨任务的文件依赖问题。

示例代码：

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.google.cloud.hooks.gcs import GCSHook
from datetime import datetime
import tempfile

def generate_and_upload():
    # 同一个任务内使用临时文件，不会被跨任务清理
    with tempfile.NamedTemporaryFile(mode='w', suffix='.csv', delete=False) as f:
        f.write("col1,col2\nval1,val2")
        temp_path = f.name
    
    # 上传到GCS
    gcs_hook = GCSHook()
    gcs_hook.upload(
        bucket_name='your-bucket-name',
        object_name='path/to/abc.csv',
        filename=temp_path
    )

with DAG(
    dag_id='generate_and_upload',
    start_date=datetime(2024,1,1),
    schedule_interval=None,
    catchup=False
) as dag:
    task = PythonOperator(
        task_id='generate_and_upload',
        python_callable=generate_and_upload
    )

总结 #

• 小文件优先用XCom传递，无本地文件依赖
• 大文件用共享存储路径/home/airflow/gcs/data/，解决跨worker节点访问问题
• 逻辑简单场景直接合并任务，最省心

内容的提问来源于stack exchange，提问作者p.magalhaes