Airflow 默认情况下不会自动删除任务创建的文件，但可以使用 PythonOperator 中的 provide_context 参数来将上下文变量传递给操作函数，并在操作函数中使用 task_instance 删除文件。示例代码如下：

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime
import os

def clean_up_files(**context):
    # get the file path from the task instance
    file_path = context['task_instance'].xcom_pull(task_ids='generate_file_task')
    # delete the file
    os.remove(file_path)

def generate_file():
    file_name = 'example.txt'
    file_path = os.path.join('/tmp', file_name)
    with open(file_path, 'w') as f:
        f.write('hello world')
    return file_path

with DAG('file_cleanup_dag', start_date=datetime(2021, 1, 1), schedule_interval=None) as dag:

    generate_file_task = PythonOperator(
        task_id='generate_file_task',
        python_callable=generate_file
    )

    clean_up_files_task = PythonOperator(
        task_id='clean_up_files_task',
        python_callable=clean_up_files,
        provide_context=True
    )

    generate_file_task >> clean_up_files_task

上述代码演示了一个 Airflow DAG，它生成一个文件并删除它。在 generate_file 函数中，它通过 os.path.join 将文件名和路径组合在一起，并使用 open 函数创建一个新文件，并将“hello world”写入该文件。接下来，在 clean_up_files 函数中，我们使用 task_instance 的 xcom_pull 方法来获取 generate_file_task 生成的文件路径，并使用 os.remove 删除文件。在 clean_up_files_task 中，我们将 provide_context 参数设置为 True，以便可以在 clean_up_files 函数中访问 task_instance。最后，我们将 generate_file_task 和 clean_up_files_task 连接起来，以便在 generate_file_task 执行完毕后立即执行 clean_up_files_task。