使用PythonOperator实现动态参数传递和配置文件读取

PythonOperator是Apache Airflow中常用的operator之一，用于将Python函数作为task运行。PythonOperator可以用于传递动态参数以及读取配置文件，提高代码的灵活性和可配置性。

首先，我们需要安装Apache Airflow，并导入相关的模块：

!pip install apache-airflow
import os
from datetime import datetime
from airflow import DAG
from airflow.operators.python_operator import PythonOperator

接下来，我们定义一个Python函数，用于使用动态参数并读取配置文件：

def dynamic_args_and_config(**kwargs):
    # 传递动态参数
    dynamic_param = kwargs["dag_run"].conf.get("param")

    # 读取配置文件
    config_file_path = kwargs["task_instance"].xcom_pull(task_ids="read_config_file")
    with open(config_file_path, "r") as f:
        config = f.read()

    print(f"Dynamic Parameter: {dynamic_param}")
    print(f"Config File Content: {config}")

然后，我们定义一个DAG，包含两个task，一个用于读取配置文件，另一个用于执行动态参数和配置文件的函数：

default_args = {
    "owner": "airflow",
    "start_date": datetime(2022, 1, 1),
}

dag = DAG(
    "dynamic_args_and_config_example",
    default_args=default_args,
    schedule_interval=None,  # 仅手动触发
)

def read_config_file():
    config_file_path = "/path/to/config_file.txt"
    return config_file_path

read_config_file_task = PythonOperator(
    task_id="read_config_file",
    python_callable=read_config_file,
    dag=dag,
)

dynamic_args_and_config_task = PythonOperator(
    task_id="dynamic_args_and_config",
    python_callable=dynamic_args_and_config,
    provide_context=True,  # 传递上下文信息
    dag=dag,
)

read_config_file_task >> dynamic_args_and_config_task

在上述代码中，我们首先定义了一个默认配置default_args，用于指定DAG的所有task共享的默认参数。然后，我们创建了一个DAG，并指定了DAG的名称、默认参数、以及调度间隔（在此例中为None，即仅手动触发）。接下来，我们定义了两个task， 个task用于从配置文件中读取配置，第二个task用于执行参数和配置文件的函数。我们通过将两个task的实例通过“>>”符号连接起来，构建task之间的依赖关系。

最后，我们可以通过命令行启动Airflow的调度器和Web服务器，然后手动触发DAG的执行：

airflow scheduler
airflow webserver

在Airflow的Web界面中，找到并点击我们创建的DAG，然后点击"Trigger DAG"按钮触发DAG的执行。当DAG执行完成后，我们可以在日志中看到动态参数和配置文件的内容。

总结：

使用PythonOperator可以方便地实现动态参数传递和配置文件读取。通过传递上下文信息，我们可以在任务中使用动态参数。同时，通过使用XCom机制，我们可以在任务之间传递数据，从而实现配置文件的读取。利用Apache Airflow的调度和监控功能，我们可以更好地管理和运行任务，提高任务的灵活性和可配置性。