AirflowPythonOperator：使用Python函数发送HTTP请求

Airflow的PythonOperator是Airflow的一个Task类，用于执行Python函数作为一个任务。我们可以使用Python函数发送HTTP请求来获取数据、执行操作等。

在使用PythonOperator发送HTTP请求之前，我们需要确保Airflow环境中已经安装了相应的库，例如requests库。可以通过运行pip install requests来安装。

接下来，我们可以定义一个Python函数来发送HTTP请求。下面是一个简单的例子：

import requests

def send_http_request():
    url = "https://api.example.com/data"
    headers = {
        "Content-Type": "application/json",
        "Authorization": "Bearer <your_token>"
    }
    
    try:
        response = requests.get(url, headers=headers)
        response.raise_for_status()  # 如果请求失败，会抛出一个异常
        data = response.json()
        return data
    except requests.exceptions.HTTPError as errh:
        print("HTTP Error:", errh)
    except requests.exceptions.ConnectionError as errc:
        print("Error Connecting:", errc)
    except requests.exceptions.Timeout as errt:
        print("Timeout Error:", errt)
    except requests.exceptions.RequestException as err:
        print("Error:", err)

在上面的代码中，我们定义了一个send_http_request函数，它发送一个GET请求到https://api.example.com/data URL。我们还设置了请求头部包含Content-Type和Authorization字段。

在发送请求之后，我们根据响应的状态码来处理不同的错误情况。如果请求成功，我们可以通过response.json()来获取响应的JSON数据，并返回数据以供后续使用。

接下来，我们可以在Airflow的DAG中使用PythonOperator来执行这个函数作为一个任务：

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

default_args = {
    'start_date': datetime(2021, 1, 1),
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'http_request_dag',
    default_args=default_args,
    schedule_interval='@daily',
)

http_request_task = PythonOperator(
    task_id='send_http_request',
    python_callable=send_http_request,
    dag=dag,
)

http_request_task

在上面的代码中，我们创建了一个DAG，并定义了一个send_http_request任务。我们设置了任务的重试次数为3次，并在每次重试之间等待了5分钟。任务的调度间隔为每天一次。

通过在DAG中添加其他的任务，例如数据处理、数据可视化等任务，我们可以构建一个完整的工作流程。

总结：Airflow的PythonOperator提供了一种在Airflow中执行Python函数作为任务的方法。我们可以使用Python函数发送HTTP请求来获取数据、执行操作等。在DAG中，通过将PythonOperator实例化为一个任务，可以将这个函数作为一个具体的任务来执行。加上其他的任务，可以构建一个完整的工作流程。