airflow dag 网络接口请求，并且把请求结果写入DB的demo。

### 回答1：
这里有一个Airflow的DAG示例，它从一个API接口请求数据，然后将结果写入数据库中。

首先，在您的Airflow安装中，您需要安装requests和pymysql库，因为它们将用于从API获取数据和将数据写入MySQL数据库。

import requests
import json
import pymysql
from datetime import datetime
from airflow import DAG
from airflow.operators.python_operator import PythonOperator

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2023, 2, 16),
    'depends_on_past': False,
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1
}

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

def get_api_data():
    url = 'https://api.example.com/data'
    headers = {'Content-Type': 'application/json'}
    response = requests.get(url, headers=headers)
    data = json.loads(response.content)
    return data

def write_to_db():
    data = get_api_data()
    conn = pymysql.connect(
        host='localhost',
        user='username',
        password='password',
        db='database'
    )
    cursor = conn.cursor()
    for row in data:
        # assuming the data has "id", "name", "value" keys
        sql = f"INSERT INTO mytable (id, name, value) VALUES ({row['id']}, '{row['name']}', {row['value']})"
        cursor.execute(sql)
    conn.commit()
    conn.close()

get_data_task = PythonOperator(
    task_id='get_api_data',
    python_callable=get_api_data,
    dag=dag
)

write_to_db_task = PythonOperator(
    task_id='write_to_db',
    python_callable=write_to_db,
    dag=dag
)

get_data_task >> write_to_db_task

这个DAG包括两个任务：`get_api_data`和`write_to_db`。`get_api_data`任务从API获取数据，`write_to_db`任务将数据写入MySQL数据库中。这些任务是通过PythonOperator运行的，其任务ID分别为`get_api_data`和`write_to_db`。

在这个示例中，我们假设API返回的数据有"id"，"name"和"value"这些键。您需要根据您的API数据结构进行相应的修改。并且，请确保在Airflow中配置正确的MySQL数据库连接信息。

### 回答2：
Airflow DAG是一个用于定义、调度和监控工作流的平台。它可以使用Python编写，支持各种任务和操作。

要创建一个Airflow DAG，首先需要导入相关库和模块，然后定义一个DAG对象，设置DAG的名称、描述、调度时间和默认参数。接下来，可以定义DAG的任务，其中一个任务可以是网络接口请求的操作。

为了进行网络接口请求，我们可以使用Python的requests库。在DAG中的任务函数中，使用requests发送网络请求，并获取返回的结果。

例如，我们可以创建一个名为"network_request"的任务函数，其中执行网络接口请求的操作。请求的URL、方法、参数等可以根据实际需求进行设置。

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

def network_request():
    url = "https://example.com/api/endpoint"
    method = "GET"
    params = {"param1": "value1", "param2": "value2"}
    response = requests.request(method, url, params=params)
    # 将请求结果写入数据库
    write_to_db(response.text)

def write_to_db(response):
    # 将请求结果写入数据库的代码
    pass

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2021, 1, 1),
    # 其他默认参数
}

dag = DAG(
    'demo',
    description='网络接口请求并写入数据库的示例DAG',
    schedule_interval='@daily',
    default_args=default_args,
    catchup=False
)

network_task = PythonOperator(
    task_id='network_request_task',
    python_callable=network_request,
    dag=dag
)

network_task

以上是一个演示如何在Airflow DAG中进行网络接口请求并将结果写入数据库的简单示例。在实际应用中，需要根据具体需求进行适当的配置和修改。

### 回答3：
Airflow是一个用于调度和监控任务的开源平台。在Airflow中，DAG（Directed Acyclic Graph）是任务的有向无环图，表示任务之间的依赖关系。

要实现Airflow DAG中的网络接口请求，可以使用Python中的requests库来发送HTTP请求。首先，在DAG中定义一个Operator来执行HTTP请求，并将结果存储到数据库中。

首先，需要安装requests库：

pip install requests

然后，在Airflow中创建一个自定义Operator，代码示例如下：

from airflow.models import BaseOperator
from airflow.utils.decorators import apply_defaults
import requests

class NetworkOperator(BaseOperator):
    @apply_defaults
    def __init__(self, url, db_conn, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.url = url
        self.db_conn = db_conn

    def execute(self, context):
        try:
            # 发送网络请求
            response = requests.get(self.url)

            # 将请求结果写入数据库
            # 使用self.db_conn执行写入操作
            # 例如：self.db_conn.execute("INSERT INTO table_name (column1, column2) VALUES (%s, %s)", (response.json()["data1"], response.json()["data2"]))

            # 如果请求成功返回True，否则返回False
            return response.status_code == 200
        except Exception as e:
            # 捕获任何异常，并打印错误消息
            self.log.error(f"请求失败：{str(e)}")
            return False

在该自定义Operator中，我们使用requests库发送GET请求，并将请求结果写入数据库（使用传入的db_conn连接执行写入操作）。

在Airflow DAG中，可以使用该自定义Operator来执行网络接口请求并将结果写入数据库，示例代码如下：

from airflow import DAG
from datetime import datetime
from network_operator import NetworkOperator

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

with DAG('network_dag', default_args=default_args, schedule_interval='@daily') as dag:
    t1 = NetworkOperator(
        task_id='network_request',
        url='https://example.com/api',
        db_conn=db_conn  # 替换为实际的数据库连接对象
    )

在以上示例中，我们定义了一个名为network_dag的DAG，该DAG将每天执行一次网络请求任务。任务使用NetworkOperator来执行网络请求，并传入URL和数据库连接对象。

这样，当DAG运行时，Airflow会按照定义的调度策略执行网络接口请求，并将请求结果写入数据库中。查看数据库中的数据，可以使用数据库连接对象进行查询操作。