如何通过Tasks迭代遍历REST端点实现分页轮询获取数据？

针对你需要持续分页拉取REST接口数据、无结果时轮询的需求，完全不用做长时间运行的DAG或者依赖全局变量，这里给你一套更合理的设计方案：

一、核心思路：拆分短任务，用外部状态存储替代DAG变量 #

把整个流程拆成单次请求+状态更新的短任务，用数据库、缓存这类外部存储来记录当前页码和轮询状态，每个任务只做一次请求，跑完就结束，完美避开长运行DAG的问题。

具体步骤：

• 初始化：第一次运行前，在状态存储里存好起始页码（比如1）和轮询标记（默认false，表示还没到需要轮询的阶段）。
• 每次任务执行：
  1. 从状态存储里读当前页码和轮询状态。
  2. 发请求GET /api/search?page={当前页码}。
  3. 如果返回有数据：
     • 处理数据（入库、解析等操作）。
     • 页码加1，把轮询标记设为false，更新到状态存储。
  4. 如果返回无数据：
     • 要是第一次碰到无数据，就把轮询标记改成true，页码不动；要是已经在轮询了，就保持当前状态不变。
• 任务调度：用调度器（比如Airflow、Prefect）按短周期触发任务（比如每5分钟一次），直到你设定的终止条件满足（比如拉取到指定时间范围的数据）。

二、为啥不用带页码变量的长DAG？ #

文档说要避免长运行DAG和全局变量，原因很实在：

• 长DAG占用资源多，一旦中断就得从头重试，容错性极差；而且调度器持续盯着长时间运行的任务，压力也会很大。
• DAG里的全局变量（比如Airflow的Variable）是共享的，要是多个任务实例同时运行，很容易出现页码被同时修改的竞争问题，导致数据拉取混乱。

用外部状态存储的话，每个任务都是短平快的单次执行，就算失败了，下次启动直接从上次记录的状态继续；而且状态存储可以加锁，完全不用担心并发冲突问题。

三、实际实现示例（以Airflow为例） #

假设用Airflow调度，用SQLite存储状态，代码大概是这样：

1. 先创建状态表：

CREATE TABLE IF NOT EXISTS api_poll_state (
    task_id TEXT PRIMARY KEY,
    current_page INTEGER NOT NULL,
    is_polling BOOLEAN NOT NULL DEFAULT false
);

2. 编写Python任务逻辑：

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
import requests
import sqlite3

def fetch_and_process_data():
    # 连接状态数据库
    conn = sqlite3.connect('/opt/airflow/state.db')
    cursor = conn.cursor()
    
    # 获取当前状态，无记录则初始化
    cursor.execute("SELECT current_page, is_polling FROM api_poll_state WHERE task_id='search_api_task'")
    result = cursor.fetchone()
    if not result:
        current_page = 1
        is_polling = False
        cursor.execute("INSERT INTO api_poll_state VALUES ('search_api_task', ?, ?)", (current_page, is_polling))
    else:
        current_page, is_polling = result
    
    # 请求接口并处理异常
    try:
        response = requests.get(f"https://your-api-domain.com/api/search?page={current_page}")
        response.raise_for_status()  # 捕获HTTP错误
        data = response.json()
    except requests.exceptions.RequestException as e:
        print(f"Request failed: {e}")
        conn.close()
        return  # 请求失败不更新状态
    
    if data.get('results'):
        # 处理数据，示例为打印，实际可写入业务数据库
        print(f"Processed page {current_page}: {len(data['results'])} items")
        # 更新状态：页码+1，退出轮询
        cursor.execute("UPDATE api_poll_state SET current_page=?, is_polling=? WHERE task_id='search_api_task'", 
                       (current_page + 1, False))
    else:
        # 无数据，进入或保持轮询状态
        cursor.execute("UPDATE api_poll_state SET is_polling=? WHERE task_id='search_api_task'", (True,))
    
    conn.commit()
    conn.close()

# 定义DAG，每5分钟触发一次
with DAG(
    dag_id='search_api_polling',
    schedule_interval='*/5 * * * *',
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=['api', 'polling']
) as dag:
    fetch_task = PythonOperator(
        task_id='fetch_and_process',
        python_callable=fetch_and_process_data
    )

这个方案里，DAG每次只运行几分钟就结束，完全符合文档里的最佳实践，也不用依赖全局变量。

四、额外注意点 #

• 接口限流：要是接口有QPS限制，记得在任务里加延迟或者重试机制，避免被接口封禁。
• 状态锁：如果可能有多个任务实例并发运行，给状态表加行级锁（比如SQLite用BEGIN EXCLUSIVE），防止页码被重复修改。
• 异常处理：网络错误、接口返回异常时，不要更新状态，直接重试即可，避免丢失当前页码。

内容的提问来源于stack exchange，提问作者Jan Martin