airflow如何同时生成不同目录的多个文件

Apache Airflow 是一个强大的任务调度系统，主要用于自动化数据处理工作流程。如果你想要使用 Airflow 同时生成不同目录下的多个文件，你可以通过编写一个自定义Operator来实现这一需求。这里是一个基本步骤：

1. **创建DAG**（Directed Acyclic Graph，有向无环图），这是Airflow的基本工作单元，代表了一个操作序列。

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

default_args = {
    'owner': 'your_name',
    'start_date': datetime(2023, 1, 1),
    'retries': 1,
    'retry_delay': timedelta(minutes=5)
}

dag = DAG(
    dag_id='generate_files',
    default_args=default_args,
    schedule_interval=timedelta(hours=1),  # 每小时运行一次
    catchup=False  # 防止重复运行
)

2. **编写Python函数**，这个函数会在每次DAG运行时被执行，并用于生成文件。你可以使用`os.makedirs()`创建目录，然后使用文件操作库（如`open()`和`write()`）来创建文件。

import os
import time
from airflow.models import Variable

def generate_files():
    base_dir = Variable.get('base_directory')  # 获取变量存储的基目录
    for i in range(3):  # 示例：生成3个文件
        subdir_name = f'subdir_{i}'
        subdir_path = os.path.join(base_dir, subdir_name)
        os.makedirs(subdir_path, exist_ok=True)  # 创建子目录，如果已存在则忽略

        file_name = f'file_{time.strftime("%Y%m%d%H%M%S")}.txt'
        file_path = os.path.join(subdir_path, file_name)

        with open(file_path, 'w') as f:
            f.write(f'This is a file generated at {time.ctime()}')

generate_files_task = PythonOperator(
    task_id='generate_files_task',
    python_callable=generate_files,
    provide_context=True,
    dag=dag
)

3. **配置变量**：为了在DAG运行时提供不同目录路径，可以使用Airflow的Variable功能设置一个变量。在Airflow UI中创建一个名为`base_directory`的变量，输入你希望生成文件的根目录。

4. **添加到DAG**：最后，在DAG中添加刚刚创建的任务。

记得将上述代码保存为一个.py文件，然后在Airflow的`dags`目录下配置好。运行DAG时，Airflow会按照预定的时间间隔执行`generate_files_task`，并生成指定目录下的文件。