DuckDB实战：HTTPS/API数据接入进阶——认证、分页与生产级管道

DuckDB实战：HTTPS/API数据接入进阶——认证、分页与生产级管道

深入掌握 DuckDB httpfs 扩展的高级用法：Bearer Token 认证、分页 API 处理、流式读取大 JSON、以及基于时间戳的增量拉取，构建生产级零中间件数据管道。

引言

架构图

图：DuckDB httpfs 生产级 API 接入架构——从认证到增量拉取的全流程

在上一篇文章中，我们介绍了 DuckDB httpfs 扩展的基础用法——直接读取远程 JSON 和 CSV 文件。但在真实的生产环境中，API 数据接入远不止"一条 SQL 读到底"那么简单。你通常会遇到以下挑战：

API 认证：Bearer Token、API Key、OAuth 2.0
分页处理：API 返回大量数据，需要逐页拉取并合并
大文件流式读取：响应 JSON 过大，内存溢出
增量拉取：只获取变更数据，避免全量重复同步
错误重试：网络抖动、速率限制（Rate Limit）

本文将通过一个电商订单系统 API 接入的真实场景，系统性地解决上述问题，帮你构建生产级 DuckDB httpfs 数据管道。

一、环境准备与扩展加载

首先确保 DuckDB 版本 >= 1.0.0，并加载 httpfs 扩展：

INSTALL httpfs;
LOAD httpfs;

-- 验证扩展已加载
SELECT name, version FROM installed_extensions WHERE name = 'httpfs';

同时设置超时和并发参数，这对生产环境很重要：

-- 设置 HTTP 超时（秒）
SET http_timeout = 30;

-- 设置并发连接数
SET http_max_connections = 10;

-- 查看当前配置
PRAGMA http_timeout;
PRAGMA http_max_connections;

终端输出

图：加载 httpfs 扩展并设置生产参数

大多数生产 API 都需要认证。DuckDB httpfs 支持通过 headers 参数传递自定义 HTTP 头。

2.1 Bearer Token 认证

假设你的 API 需要 Bearer Token 认证：

-- 方式一：直接在 URL 查询中使用 headers 参数
SELECT * FROM read_json_auto(
    'https://api.example.com/v2/orders',
    headers => {'Authorization': 'Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...'}
);

-- 方式二：将认证信息存入 DuckDB credentials（避免硬编码）
CREATE CREDENTIAL duckdb_api_creds (
    'Authorization' VALUE 'Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...'
);

-- 使用凭证
SELECT * FROM read_json_auto(
    'https://api.example.com/v2/orders',
    credential => 'duckdb_api_creds'
);

2.2 API Key 认证

有些 API 使用 API Key 作为查询参数或 Header：

-- API Key 作为 Header
SELECT * FROM read_csv_auto(
    'https://api.example.com/v1/products',
    headers => {'X-API-Key': 'pk_live_abc123xyz'}
);

-- API Key 作为查询参数
SELECT * FROM read_json_auto(
    'https://api.example.com/v1/products?api_key=pk_live_abc123xyz'
);

2.3 基础认证（Basic Auth）

SELECT * FROM read_json_auto(
    'https://api.example.com/v1/secure-data',
    headers => {
        'Authorization': 'Basic ' || encode(base64('username:password'), 'UTF8')
    }
);

安全提示：生产环境中建议使用 DuckDB 的 Credential 系统或环境变量管理密钥，不要将 Token 硬编码在 SQL 中。

三、分页 API 处理：从单页到全量数据

当 API 返回数据量较大时，通常会采用分页机制。DuckDB 可以通过循环 + 合并的方式处理分页。

3.1 理解分页参数

典型的分页 API 使用 page 和 limit 参数：

GET https://api.example.com/v2/orders?page=1&limit=100
GET https://api.example.com/v2/orders?page=2&limit=100
GET https://api.example.com/v2/orders?page=3&limit=100
...

3.2 使用 Python 循环拉取全量数据

虽然 DuckDB SQL 本身不支持循环，但我们可以用 Python 脚本结合 DuckDB 完成分页拉取：

import duckdb
import requests
import json

DB_PATH = "ecommerce.duckdb"
API_BASE = "https://api.example.com/v2"
TOKEN = "Bearer your_token_here"
PAGE_SIZE = 100

con = duckdb.connect(DB_PATH)
con.execute("INSTALL httpfs; LOAD httpfs;")

all_orders = []
page = 1

while True:
    resp = requests.get(
        f"{API_BASE}/orders",
        params={"page": page, "limit": PAGE_SIZE},
        headers={"Authorization": TOKEN}
    )
    data = resp.json()
    
    if not data.get("items"):
        break
    
    all_orders.extend(data["items"])
    total = data.get("total", 0)
    print(f"Page {page}: got {len(data['items'])} records, total so far: {len(all_orders)}")
    
    if len(all_orders) >= total:
        break
    page += 1

# 写入 DuckDB
con.execute("CREATE TABLE IF NOT EXISTS orders (order_id VARCHAR, customer_id VARCHAR, product VARCHAR, quantity INTEGER, amount DECIMAL, order_date TIMESTAMP, status VARCHAR)")
con.execute("DELETE FROM orders")  # 全量覆盖
con.execute("INSERT INTO orders SELECT * FROM (VALUES " + ",".join(
    [f"({json.dumps(o['order_id'])}, {json.dumps(o['customer_id'])}, {json.dumps(o['product'])}, {o['quantity']}, {o['amount']}, {json.dumps(o['order_date'])}, {json.dumps(o['status'])})" for o in all_orders[:100]]
) + ")")  # 实际使用 con.execute(\"INSERT INTO orders SELECT * FROM all_orders_df\")

print(f"Total records loaded: {len(all_orders)}")
con.close()

3.3 使用 DuckDB 内置函数处理嵌套分页

对于响应中包含分页信息的 API，可以用 DuckDB 递归 CTE 处理：

-- 假设第一页数据已加载
CREATE TABLE api_page_1 AS
SELECT * FROM read_json_auto(
    'https://api.example.com/v2/orders?page=1&limit=100',
    headers => {'Authorization': 'Bearer your_token'}
);

-- 递归获取后续页面（DuckDB 1.1+ 支持 recursive with）
WITH RECURSIVE fetched_pages AS (
    SELECT 1 AS page_num, * FROM api_page_1
    UNION ALL
    SELECT fp.page_num + 1, 
           * FROM read_json_auto(
               'https://api.example.com/v2/orders?page=' || (fp.page_num + 1) || '&limit=100',
               headers => {'Authorization': 'Bearer your_token'}
           ) AS t
    WHERE fp.page_num < 5  -- 设置最大页数限制
)
SELECT * FROM fetched_pages;

注意：递归 CTE 方式适合页数较少的场景。对于大量分页，推荐 Python 循环方案。

四、流式读取大 JSON 响应

当 API 返回的 JSON 非常大（数百MB甚至GB级）时，直接 read_json_auto 可能导致内存溢出。DuckDB 提供了流式读取能力。

4.1 使用 `streaming` 选项

-- 流式读取大型 JSON（DuckDB 1.0+）
CREATE TABLE large_orders AS
SELECT * FROM read_json_auto(
    'https://api.example.com/v2/orders?limit=50000',
    streaming => true,
    headers => {'Authorization': 'Bearer your_token'}
);

4.2 分块处理 + 增量写入

对于超大数据集，可以分块处理并增量写入：

import duckdb
import requests

con = duckdb.connect("ecommerce.duckdb")
con.execute("INSTALL httpfs; LOAD httpfs;")

# 创建表结构
con.execute("""
    CREATE TABLE IF NOT EXISTS orders_batch (
        order_id VARCHAR,
        customer_id VARCHAR,
        product VARCHAR,
        quantity INTEGER,
        amount DECIMAL(10,2),
        order_date TIMESTAMP,
        status VARCHAR
    )
""")

# 分批拉取并写入
batch_size = 10000
offset = 0

while True:
    resp = requests.get(
        "https://api.example.com/v2/orders",
        params={"limit": batch_size, "offset": offset},
        headers={"Authorization": "Bearer your_token"}
    )
    data = resp.json()
    
    if not data.get("items"):
        break
    
    # 将 JSON 写入 DuckDB
    df = con.sql(f"""
        SELECT * FROM read_json_auto(
            'https://api.example.com/v2/orders?limit={batch_size}&offset={offset}',
            streaming => true,
            headers => {{'Authorization': 'Bearer your_token'}}
        )
    """).fetchdf()
    
    con.execute("INSERT INTO orders_batch SELECT * FROM df")
    print(f"Batch {offset//batch_size + 1}: inserted {len(df)} rows")
    
    offset += batch_size
    if len(df) < batch_size:
        break

# 统计
result = con.execute("SELECT COUNT(*) FROM orders_batch").fetchone()
print(f"Total: {result[0]} records")
con.close()

终端输出

图：流式读取大 JSON 并分批写入 DuckDB

五、基于时间戳的增量拉取

全量拉取效率低且浪费资源。生产环境通常采用增量拉取——只获取上次同步以来的变更数据。

5.1 增量拉取的基本思路

1. 记录上次同步的时间戳
2. 查询 API 获取 updatedAt > 上次时间戳 的数据
3. 使用 UPSERT 合并到本地表
4. 更新同步时间戳

5.2 实现增量拉取

-- 创建增量同步状态表
CREATE TABLE IF NOT EXISTS sync_state (
    source VARCHAR PRIMARY KEY,
    last_synced_at TIMESTAMP
);

-- 初始化同步状态
INSERT INTO sync_state (source, last_synced_at)
VALUES ('orders_api', '2026-01-01 00:00:00')
ON CONFLICT (source) DO NOTHING;

-- 创建目标表（支持 UPSERT）
CREATE TABLE IF NOT EXISTS orders_incremental (
    order_id VARCHAR PRIMARY KEY,
    customer_id VARCHAR,
    product VARCHAR,
    quantity INTEGER,
    amount DECIMAL(10,2),
    order_date TIMESTAMP,
    status VARCHAR,
    updated_at TIMESTAMP,
    synced_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

import duckdb
from datetime import datetime, timedelta

con = duckdb.connect("ecommerce.duckdb")
con.execute("INSTALL httpfs; LOAD httpfs;")

# 读取上次同步时间
last_sync = con.execute("SELECT last_synced_at FROM sync_state WHERE source = 'orders_api'").fetchone()[0]
print(f"Last sync: {last_sync}")

# 增量拉取（假设 API 支持 ?since= 参数）
new_data = con.execute(f"""
    SELECT * FROM read_json_auto(
        'https://api.example.com/v2/orders?since={last_sync}',
        headers => {{'Authorization': 'Bearer your_token'}}
    )
""").fetchdf()

print(f"New records: {len(new_data)}")

# UPSERT 合并
if len(new_data) > 0:
    con.execute("CREATE TEMP TABLE new_batch AS SELECT * FROM new_data")
    con.execute("""
        INSERT INTO orders_incremental
        SELECT * FROM new_batch
        ON CONFLICT (order_id) DO UPDATE SET
            customer_id = EXCLUDED.customer_id,
            product = EXCLUDED.product,
            quantity = EXCLUDED.quantity,
            amount = EXCLUDED.amount,
            order_date = EXCLUDED.order_date,
            status = EXCLUDED.status,
            updated_at = EXCLUDED.updated_at,
            synced_at = CURRENT_TIMESTAMP
    """)

# 更新同步时间
new_sync_time = con.execute("SELECT MAX(updated_at) FROM new_batch").fetchone()[0]
con.execute(f"""
    UPDATE sync_state 
    SET last_synced_at = '{new_sync_time}'
    WHERE source = 'orders_api'
""")

# 查看增量结果
result = con.execute("SELECT COUNT(*) as total, MIN(synced_at) as earliest_sync FROM orders_incremental").fetchone()
print(f"Total records: {result[0]}, Earliest sync: {result[1]}")

con.close()

5.3 定时增量同步脚本

#!/usr/bin/env python3
"""DuckDB 增量同步调度器"""
import duckdb
import schedule
import time
from datetime import datetime

def sync_orders():
    con = duckdb.connect("ecommerce.duckdb")
    con.execute("INSTALL httpfs; LOAD httpfs;")
    
    last_sync = con.execute(
        "SELECT last_synced_at FROM sync_state WHERE source = 'orders_api'"
    ).fetchone()[0]
    
    # 拉取增量数据
    df = con.execute(f"""
        SELECT * FROM read_json_auto(
            'https://api.example.com/v2/orders?since={last_sync}',
            headers => {{'Authorization': 'Bearer your_token'}}
        )
    """).fetchdf()
    
    if len(df) > 0:
        con.execute("CREATE TEMP TABLE batch AS SELECT * FROM df")
        con.execute("""
            INSERT INTO orders_incremental
            SELECT * FROM batch
            ON CONFLICT (order_id) DO UPDATE SET
                status = EXCLUDED.status,
                updated_at = EXCLUDED.updated_at,
                synced_at = CURRENT_TIMESTAMP
        """)
        new_time = df['updated_at'].max()
        con.execute(f"UPDATE sync_state SET last_synced_at = '{new_time}' WHERE source = 'orders_api'")
        print(f"[{datetime.now()}] Synced {len(df)} new records")
    
    con.close()

# 每5分钟同步一次
schedule.every(5).minutes.do(sync_orders)

while True:
    schedule.run_pending()
    time.sleep(1)

六、错误处理与重试机制

生产环境必须考虑网络异常和速率限制。

6.1 使用 DuckDB 内置重试

-- DuckDB 1.1+ 支持自动重试
SELECT * FROM read_json_auto(
    'https://api.example.com/v2/orders',
    headers => {'Authorization': 'Bearer your_token'},
    retry => true,
    retry_max => 3,
    retry_delay_ms => 1000
);

6.2 Python 端重试（更灵活）

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# 创建带重试的 session
session = requests.Session()
retry = Retry(
    total=3,
    backoff_factor=1,
    status_forcelist=[429, 500, 502, 503, 504]
)
adapter = HTTPAdapter(max_retries=retry)
session.mount('https://', adapter)

# 使用重试 session 拉取数据
resp = session.get(
    'https://api.example.com/v2/orders',
    headers={'Authorization': 'Bearer your_token'},
    params={'page': 1, 'limit': 100}
)

6.3 速率限制处理

import time

def fetch_with_rate_limit(url, headers, max_retries=5):
    for attempt in range(max_retries):
        resp = requests.get(url, headers=headers)
        
        if resp.status_code == 200:
            return resp.json()
        elif resp.status_code == 429:  # Rate limited
            retry_after = int(resp.headers.get('Retry-After', 2 ** attempt))
            print(f"Rate limited. Waiting {retry_after}s...")
            time.sleep(retry_after)
        else:
            raise Exception(f"HTTP {resp.status_code}")
    
    raise Exception(f"Failed after {max_retries} retries")

七、完整生产管道示例

将以上所有技术整合到一个完整的数据管道：

#!/usr/bin/env python3
"""
DuckDB 生产级 API 数据管道
功能：认证 + 分页 + 增量 + 重试 + UPSERT
"""
import duckdb
import requests
import json
import time
from datetime import datetime
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

class DuckDBAPIPipeline:
    def __init__(self, db_path, api_base, token):
        self.db_path = db_path
        self.api_base = api_base
        self.token = token
        self.con = duckdb.connect(db_path)
        self._setup()
        
        # 带重试的 HTTP session
        self.session = requests.Session()
        retry = Retry(total=3, backoff_factor=1, 
                      status_forcelist=[429, 500, 502, 503, 504])
        self.session.mount('https://', HTTPAdapter(max_retries=retry))
    
    def _setup(self):
        """初始化表结构和扩展"""
        self.con.execute("INSTALL httpfs; LOAD httpfs;")
        
        self.con.execute("""
            CREATE TABLE IF NOT EXISTS orders (
                order_id VARCHAR PRIMARY KEY,
                customer_id VARCHAR,
                product VARCHAR,
                quantity INTEGER,
                amount DECIMAL(10,2),
                order_date TIMESTAMP,
                status VARCHAR,
                updated_at TIMESTAMP,
                synced_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        
        self.con.execute("""
            CREATE TABLE IF NOT EXISTS sync_state (
                source VARCHAR PRIMARY KEY,
                last_synced_at TIMESTAMP,
                last_sync_status VARCHAR,
                record_count INTEGER
            )
        """)
    
    def incremental_sync(self):
        """增量同步主流程"""
        # 读取上次同步时间
        row = self.con.execute(
            "SELECT last_synced_at FROM sync_state WHERE source = 'orders'"
        ).fetchone()
        
        if row:
            last_sync = row[0]
        else:
            last_sync = "2026-01-01 00:00:00"
        
        print(f"[{datetime.now()}] Starting incremental sync from {last_sync}")
        
        # 分页拉取增量数据
        all_records = []
        page = 1
        page_size = 100
        
        while True:
            resp = self.session.get(
                f"{self.api_base}/orders",
                params={"page": page, "limit": page_size, "since": last_sync},
                headers={"Authorization": self.token}
            )
            
            if resp.status_code == 429:
                wait = int(resp.headers.get('Retry-After', 5))
                print(f"  Rate limited, waiting {wait}s...")
                time.sleep(wait)
                continue
            
            data = resp.json()
            items = data.get("items", [])
            
            if not items:
                break
            
            all_records.extend(items)
            print(f"  Page {page}: got {len(items)} records (total: {len(all_records)})")
            
            if len(all_records) >= data.get("total", 0):
                break
            page += 1
        
        # UPSERT 写入
        if all_records:
            self.con.execute("CREATE TEMP TABLE batch AS SELECT * FROM (VALUES " + 
                ",".join([f"({json.dumps(r['order_id'])}, {json.dumps(r['customer_id'])}, "
                         f"{json.dumps(r['product'])}, {r['quantity']}, {r['amount']}, "
                         f"{json.dumps(r['order_date'])}, {json.dumps(r['status'])}, "
                         f"{json.dumps(r.get('updated_at', r['order_date']))})"
                         for r in all_records[:500]]) + "))")
            
            self.con.execute("""
                INSERT INTO orders
                SELECT * FROM batch
                ON CONFLICT (order_id) DO UPDATE SET
                    customer_id = EXCLUDED.customer_id,
                    product = EXCLUDED.product,
                    quantity = EXCLUDED.quantity,
                    amount = EXCLUDED.amount,
                    status = EXCLUDED.status,
                    updated_at = EXCLUDED.updated_at,
                    synced_at = CURRENT_TIMESTAMP
            """)
            
            # 更新同步状态
            new_time = max(r.get('updated_at', r['order_date']) for r in all_records)
            self.con.execute("""
                INSERT INTO sync_state (source, last_synced_at, last_sync_status, record_count)
                VALUES ('orders', ?, 'success', ?)
                ON CONFLICT (source) DO UPDATE SET
                    last_synced_at = EXCLUDED.last_synced_at,
                    last_sync_status = EXCLUDED.last_sync_status,
                    record_count = EXCLUDED.record_count
            """, [new_time, len(all_records)])
            
            print(f"[{datetime.now()}] Synced {len(all_records)} records")
        else:
            print(f"[{datetime.now()}] No new records")
        
        return len(all_records)
    
    def run_query(self, sql):
        """执行查询并返回结果"""
        return self.con.execute(sql).fetchdf()
    
    def close(self):
        self.con.close()

# 使用示例
if __name__ == "__main__":
    pipeline = DuckDBAPIPipeline(
        db_path="ecommerce.duckdb",
        api_base="https://api.example.com/v2",
        token="Bearer your_token_here"
    )
    
    try:
        # 执行增量同步
        count = pipeline.incremental_sync()
        
        # 查看同步结果
        stats = pipeline.run_query("SELECT COUNT(*) as total, MIN(synced_at) as first_sync FROM orders")
        print(f"Database stats: {stats.to_dict('records')}")
    finally:
        pipeline.close()

八、性能优化建议

优化项	配置	说明
流式读取	`streaming => true`	大 JSON 避免内存溢出
连接池	`http_max_connections`	提高并发拉取效率
超时设置	`http_timeout`	防止长时间阻塞
分页大小	`limit=100~500`	平衡内存和请求次数
增量策略	`since` 参数	避免全量重复同步
自动重试	`retry => true`	提高网络稳定性

总结

本文系统介绍了 DuckDB httpfs 扩展在生产环境中的高级用法：

API 认证：Bearer Token、API Key、Basic Auth 的多种实现方式
分页处理：Python 循环拉取 + DuckDB 递归 CTE 两种方案
流式读取：大 JSON 的 streaming 选项和分块写入
增量拉取：基于时间戳的 UPSERT 模式，避免全量重复同步
错误处理：自动重试、速率限制等待、超时控制
完整管道：整合所有技术的生产级 Python 脚本

通过这套方案，你可以构建高效、稳定、自动化的 DuckDB 数据管道，实现从外部 API 到本地分析的零中间件流程。

更多 DuckDB 实战技巧，请关注 DuckDB Lab（duckdblab.org）

📺 Watch video tutorials → Olap Studio YouTube

Subscribe for more DuckDB & AI automation tutorials

⚠️ 本站为独立社区项目，与 DuckDB 基金会及 DuckDB 官方项目无任何从属、背书或赞助关系。

"DuckDB" 是 DuckDB 基金会的注册商标，本站仅以事实描述方式使用该名称。

本站内容仅供教育与社区推广用途，不构成任何商业服务。