如何用SQLite存储Twitter趋势话题数据？完整项目实战教程

做数据分析或者做热点聚合类应用的朋友，经常需要把Twitter（现在也叫X）的趋势话题拉下来存到本地，方便后续做时间序列分析或者词频统计。这类需求如果直接上MySQL、PostgreSQL这类需要独立服务的数据库，多少有点杀鸡用牛刀，而SQLite作为单文件嵌入式数据库，天然适合这种单机采集场景：不需要安装服务、备份就是复制文件、性能对中小数据量完全够用。这篇文章就带大家完整走一遍从建表到采集入库再到查询统计的全流程。

一、趋势话题数据的特点与表结构设计

在设计表结构之前，先想清楚趋势话题数据的形态。Twitter的趋势接口（trends/place）返回的数据包含WOEID（Yahoo的地理位置编号）、趋势名称、推送地址、搜索量、是否为推广趋势等字段。趋势数据是按地区和时间两个维度变化的，比如东京和纽约在同一时刻的热门话题完全不同，同一个地区每隔几分钟抓一次，数据也会不断刷新。

针对这种特点，推荐的策略是双表设计：一张表存每次抓取的快照记录，另一张表存去重后的话题主数据。这样既能保留历史趋势轨迹，又能快速统计某个话题总共出现过多少次。建表语句如下：

-- 趋势快照表：每次抓取一行记录
CREATE TABLE IF NOT EXISTS trend_snapshots (
    snapshot_id INTEGER PRIMARY KEY AUTOINCREMENT,
    woeid INTEGER NOT NULL,             -- 地区编号，1代表全球
    captured_at TEXT NOT NULL DEFAULT (datetime('now', 'localtime')),
    raw_json TEXT                       -- 原始返回，留作备用
);

-- 话题出现记录表：关联快照，记录排名与热度
CREATE TABLE IF NOT EXISTS trend_topics (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    snapshot_id INTEGER NOT NULL,
    topic_name TEXT NOT NULL,
    tweet_volume INTEGER,               -- 推文量，可能为空
    rank_position INTEGER,              -- 当时的排名
    promoted INTEGER DEFAULT 0,         -- 是否推广话题
    FOREIGN KEY (snapshot_id) REFERENCES trend_snapshots(snapshot_id)
);

-- 索引：按地区和时间查询、按话题名统计都很快
CREATE INDEX IF NOT EXISTS idx_snapshots_woeid_time
    ON trend_snapshots(woeid, captured_at);
CREATE INDEX IF NOT EXISTS idx_topics_name
    ON trend_topics(topic_name);

这里有几个设计细节值得说明。第一，tweet_volume字段要允许为空，因为Twitter接口里不少趋势是不带推文量的，如果设成NOT NULL会导致插入失败。第二，把原始JSON也存一份到raw_json字段，虽然会多占空间，但当后续想补充新字段时（比如趋势的分类标签），可以直接从原始JSON回填，不用重新抓取。第三，时间字段用SQLite内置的datetime('now', 'localtime')生成默认值，省去了应用层传时间的麻烦。

二、用Python抓取数据并批量写入SQLite

数据入库环节推荐用Python的requests库调接口，用内置的sqlite3模块写库，不需要额外安装任何数据库驱动。下面的完整脚本实现了抓取指定地区趋势并入库的功能：

import requests
import sqlite3
import json

DB_PATH = "trends.db"

def fetch_trends(woeid, bearer_token):
    """调用Twitter趋势接口"""
    url = f"https://api.twitter.com/1.1/trends/place.json?id={woeid}"
    headers = {"Authorization": f"Bearer {bearer_token}"}
    resp = requests.get(url, headers=headers, timeout=15)
    resp.raise_for_status()
    return resp.json()[0]["trends"]

def save_trends(woeid, trends):
    """快照与话题明细一次性写入，使用事务保证一致性"""
    conn = sqlite3.connect(DB_PATH)
    try:
        cur = conn.cursor()
        cur.execute(
            "INSERT INTO trend_snapshots (woeid, raw_json) VALUES (?, ?)",
            (woeid, json.dumps(trends, ensure_ascii=False))
        )
        snapshot_id = cur.lastrowid
        rows = [
            (snapshot_id, t["name"], t.get("tweet_volume"),
             idx + 1, 1 if t.get("promoted_content") else 0)
            for idx, t in enumerate(trends)
        ]
        cur.executemany(
            "INSERT INTO trend_topics "
            "(snapshot_id, topic_name, tweet_volume, rank_position, promoted) "
            "VALUES (?, ?, ?, ?, ?)",
            rows
        )
        conn.commit()
    finally:
        conn.close()

if __name__ == "__main__":
    trends = fetch_trends(23424856, "你的BearerToken")  # 23424856为日本
    save_trends(23424856, trends)
    print(f"本次共入库 {len(trends)} 条话题")

这段代码有两个地方值得注意。首先是executemany配合事务的使用：快照和话题明细是主外键关系，必须放在同一个事务里，要么全部写入成功，要么全部回滚，避免出现孤儿记录。其次是t.get("tweet_volume")的写法，用字典的get方法可以在字段缺失时返回None，对应SQLite里的NULL，比直接取值更健壮。

实际跑起来之后，可以用crontab或者Windows任务计划每隔十分钟执行一次脚本，几天下来就能积累一份不错的历史数据。如果担心脚本重复运行产生重复快照，可以在trend_snapshots表上给captured_at加唯一索引，配合INSERT OR IGNORE语句实现幂等写入。

三、高频写入的性能优化：WAL模式与批量插入对比

采集任务跑起来之后，写入性能是绕不开的话题。SQLite默认的journal模式下，每次写入都会产生磁盘级的锁，如果多个脚本并发写同一个数据库文件，很容易碰到database is locked报错。解决办法是启用WAL（Write-Ahead Logging）模式，开启后读写可以并行，写操作先写日志文件再合并回主库，并发能力大幅提升。开启方式只需一行：

PRAGMA journal_mode = WAL;
PRAGMA synchronous = NORMAL;  -- 配合WAL使用，兼顾性能与安全
PRAGMA cache_size = -8000;    -- 使用约8MB的页面缓存

WAL模式有几个额外的好处：一是读操作不再被写操作阻塞，你可以一边采集一边用别的连接查询数据做分析；二是崩溃恢复更快，WAL日志文件记录了所有未合并的写入，数据库损坏的概率显著降低。需要注意的是WAL模式会产生-wal和-shm两个附属文件，备份时要把它们一并处理，或者先执行PRAGMA wal_checkpoint(TRUNCATE)把日志合并回主库再复制文件。

另一个性能要点是逐条插入与批量插入的差异。实测下来，不开事务逐条插入一万条记录可能需要十几秒，因为每条INSERT都隐式开启并提交一次事务，磁盘IO被反复触发；而包在同一个事务里批量提交，同样的数据量通常一秒内就能完成，差距在十倍以上。所以养成begin之后批量操作、最后统一commit的习惯，是使用SQLite最基本的性能准则。

四、常用查询统计与避坑小结

数据积累起来之后，几条高频SQL就能满足大部分分析需求。比如统计过去24小时全球最常出现的话题：

SELECT tt.topic_name,
       COUNT(*) AS appear_count,
       MAX(tt.tweet_volume) AS max_volume
FROM trend_topics tt
JOIN trend_snapshots ts ON ts.snapshot_id = tt.snapshot_id
WHERE ts.woeid = 1
  AND ts.captured_at >= datetime('now', '-1 day', 'localtime')
GROUP BY tt.topic_name
ORDER BY appear_count DESC, max_volume DESC
LIMIT 20;

再比如查某个话题的排名变化曲线，只需要按时间排序取出rank_position，导出后可以直接画折线图。查询时如果发现速度慢，先看执行计划：EXPLAIN QUERY PLAN能告诉你有没有走索引，全表扫描的话通常是索引没建对。

最后说几个常见的坑。一是并发写的问题，即使开了WAL，SQLite同一时刻也只允许一个写者，多个采集进程同时写入要考虑加超时参数sqlite3.connect(DB_PATH, timeout=30)，避免立刻报锁错误。二是字符编码，Twitter返回的话题里经常带表情符号，建库时指定PRAGMA encoding = 'UTF-8'并确保Python端不强行转码，就不会出现乱码。三是API限流，趋势接口有请求频率上限，采集间隔建议不低于十分钟，脚本里加上失败重试逻辑会更稳妥。掌握这些要点之后，用SQLite做趋势话题的采集存储基本就没什么障碍了。

SQLite趋势话题存储Twitter API修改时间：2026-09-08 14:05:22