第四节：实时数据流接入——用Tweepy搭建你的Streaming Client

好，咱们进入实战环节了。

前面几节我们聊了情绪分析的算法模型、数据清洗的套路。但有个问题一直没解决——数据从哪来？

你总不能每天手动去Twitter搜关键词，然后复制粘贴吧？那也太原始了。真正的情绪监控，必须是实时的。推文一发出，你的系统就要抓到它，分析它，然后告诉你：这条是正面还是负面。

这一节，我们就来解决这个「实时数据流」的问题。

4.1 为什么是Tweepy？

Twitter官方提供了API，但直接调用REST API有个痛点——你得不停地轮询（polling）。每5秒发一次请求，效率低，还容易被限流。

Streaming API就不一样了。它像一条水管，Twitter那边有新的推文，就顺着水管流到你这边。你只需要打开水龙头，等着接水就行。

Tweepy是Python里最成熟的Twitter API封装库。我个人用了好几年，社区活跃，文档也全。说白了，它就是帮你把那些繁琐的认证、连接、重连逻辑都包好了，你只管写业务逻辑。

核心概念： Streaming Client vs REST Client
REST：你主动去问Twitter「有新推文吗？」—— 低效，有延迟
Streaming：Twitter主动推给你「新推文来了！」—— 实时，高效

4.2 建立你的第一个Streaming Client

先装包。这个不用多说：

pip install tweepy

然后，你需要去Twitter Developer Portal申请API密钥。嗯，这一步很多人会卡住。我当初申请的时候，填了三次才通过。注意：一定要写清楚你的应用场景，别只写「测试」，写「社交媒体情绪分析研究」通过率更高。

拿到四个关键信息：

  • API Key 和 API Secret —— 你的身份标识
  • Access Token 和 Access Token Secret —— 你的访问凭证

好，开始写代码。先创建一个监听器类：

import tweepy

class TweetStreamListener(tweepy.StreamingClient):
    """
    自定义的流式监听器
    每当有新的推文匹配到过滤规则，就会触发 on_tweet 方法
    """
    def on_tweet(self, tweet):
        # 拿到推文文本
        text = tweet.text
        print(f"抓到一条推文: {text[:50]}...")
        
        # 这里你可以把数据传给后续的情绪分析模块
        # 比如调用你的情感分类模型
        
    def on_errors(self, errors):
        print(f"出错了: {errors}")
        # 我建议在这里加个日志记录，方便排查
        
    def on_connection_error(self):
        print("连接断了，尝试重连...")
        # Tweepy 会自动重连，但你可以在这里加一些自定义逻辑

然后，用你的密钥初始化客户端，并开始监听：

# 初始化客户端
bearer_token = "你的Bearer Token"
stream_client = TweetStreamListener(bearer_token)

# 添加过滤规则
# 注意：规则是持久化的，重复添加会报错
rule = tweepy.StreamRule("人工智能 OR AI OR 机器学习")
stream_client.add_rules(rule)

# 开始监听
stream_client.filter(
    tweet_fields=["lang", "created_at", "geo"],
    expansions=["author_id"]
)
小技巧： 第一次运行前，先调用 stream_client.get_rules() 看看有没有残留规则。如果有，用 stream_client.delete_rules() 清掉。我吃过这个亏，重复添加规则导致程序崩溃。

4.3 过滤规则设置——关键词、语言、地点

你不可能把Twitter上所有的推文都抓下来。那数据量太大了，而且大部分跟你无关。所以，过滤规则就是你的「筛子」。

4.3.1 关键词过滤

最基础的方式。用 StreamRule 定义你想关注的关键词：

# 单个关键词
rule1 = tweepy.StreamRule("比特币")

# 多个关键词（OR 逻辑）
rule2 = tweepy.StreamRule("比特币 OR 以太坊 OR 区块链")

# 精确短语（用引号）
rule3 = tweepy.StreamRule('"人工智能"')

# 排除某些词（用 - 号）
rule4 = tweepy.StreamRule("比特币 -广告 -推广")

嗯，这里要注意：规则有长度限制，单个规则不能超过512个字符。如果你要监控的话题很多，建议拆成多个规则。

4.3.2 语言过滤

我们做中文情绪分析，当然只关注中文推文。在 filter() 方法里加参数：

stream_client.filter(
    languages=["zh"],  # 只接收中文
    tweet_fields=["lang", "created_at"]
)

支持的常见语言代码：

语言 代码
中文（简体） zh
中文（繁体） zh-TW
英语 en
日语 ja
韩语 ko
注意： 语言过滤是在客户端做的，不是服务端。也就是说，Twitter还是会推送所有语言的推文给你，只是Tweepy帮你过滤掉了非目标语言的。如果你想节省带宽，可以在关键词里加一些中文特有的词，变相实现服务端过滤。

4.3.3 地点过滤

有时候你需要按地域分析情绪。比如，只关注北京地区的用户对某个事件的反应。

# 地点过滤需要指定地理边界框（经纬度）
# 格式：[经度1, 纬度1, 经度2, 纬度2]
# 北京的大致范围
beijing_bbox = [116.0, 39.7, 116.8, 40.2]

stream_client.filter(
    locations=beijing_bbox
)

不过说实话，地点过滤的命中率不高。因为大部分用户没有开启地理位置标签。我做过测试，只有不到5%的推文带有精确的地理位置信息。所以，如果你对地域要求很高，建议结合用户资料里的「所在地」字段做二次过滤。

4.4 处理流式数据——别让数据淹死你

数据流一旦打开，推文就像洪水一样涌进来。高峰期每秒可能几百条。你的程序如果处理不过来，就会堆积、崩溃。

我分享几个实战经验：

4.4.1 异步处理 vs 同步处理

默认情况下，on_tweet 方法是同步的。也就是说，处理完一条才能处理下一条。如果处理逻辑很慢（比如调用深度学习模型），就会阻塞整个流。

解决方案：用队列解耦

from queue import Queue
import threading

# 创建一个队列，作为缓冲区
tweet_queue = Queue(maxsize=1000)

class AsyncTweetStreamListener(tweepy.StreamingClient):
    def on_tweet(self, tweet):
        # 只做简单的入队操作，立刻返回
        tweet_queue.put(tweet)
        
    def on_errors(self, errors):
        print(f"错误: {errors}")

# 在另一个线程中处理数据
def process_tweets():
    while True:
        tweet = tweet_queue.get()
        # 这里做耗时的情绪分析
        result = analyze_sentiment(tweet.text)
        save_to_database(tweet, result)

# 启动处理线程
processor = threading.Thread(target=process_tweets, daemon=True)
processor.start()

# 主线程继续监听
stream_client.filter(tweet_fields=["text", "lang"])

这样，监听线程只管收数据，处理线程慢慢消化。队列满了怎么办？可以丢弃旧数据，或者写入磁盘临时文件。我个人习惯是设置一个合理的队列大小，超过就丢弃最早的数据——毕竟实时分析，旧数据价值不大。

4.4.2 断线重连机制

Streaming连接很不稳定。网络波动、Twitter服务器重启、API限流，都可能导致断开。

Tweepy自带重连机制，但默认参数不一定适合你。我建议手动配置：

stream_client = AsyncTweetStreamListener(bearer_token)

# 自定义重连参数
stream_client.auto_reconnect = True
stream_client.max_retries = 10          # 最大重试次数
stream_client.retry_delay = 5           # 重试间隔（秒）
stream_client.retry_time_total = 300    # 总重试时间（秒）

嗯，这里有个坑：不要无限重连。如果连续断了10次，说明你的网络或者API密钥有问题，再重连也是白费。我建议记录重连次数，超过阈值就发告警邮件。

4.4.3 数据持久化

流式数据如果不存下来，一旦程序崩溃就全丢了。我建议至少做两级存储：

  • 内存缓存：用Redis或Python的deque，存最近1000条，用于实时展示
  • 磁盘存储：每5分钟批量写入一次数据库（MongoDB或Elasticsearch）
import json
import time

class BufferedStreamListener(tweepy.StreamingClient):
    def __init__(self, bearer_token):
        super().__init__(bearer_token)
        self.buffer = []
        self.last_flush = time.time()
        self.flush_interval = 300  # 5分钟刷一次
        
    def on_tweet(self, tweet):
        # 提取关键字段
        data = {
            "id": tweet.id,
            "text": tweet.text,
            "lang": tweet.lang,
            "created_at": str(tweet.created_at),
            "author_id": tweet.author_id
        }
        self.buffer.append(data)
        
        # 检查是否需要刷盘
        if time.time() - self.last_flush > self.flush_interval:
            self.flush_to_disk()
            
    def flush_to_disk(self):
        if not self.buffer:
            return
        # 写入JSON Lines格式文件
        with open("tweets.jsonl", "a", encoding="utf-8") as f:
            for item in self.buffer:
                f.write(json.dumps(item, ensure_ascii=False) + "\n")
        self.buffer.clear()
        self.last_flush = time.time()
        print(f"已写入 {len(self.buffer)} 条数据")

4.5 本章知识体系

说了这么多，咱们用一张图来总结一下整个数据流的架构：

Twitter Streaming API Tweepy Streaming Client 过滤规则设置 关键词/语言/地点 队列 Queue maxsize=1000 实时 分析 持久化存储 JSON Lines / 数据库 情绪 分析 数据流方向：Twitter → 过滤 → 缓冲 → 实时分析 + 持久化 断线自动重连机制

这张图展示了整个数据流的走向。从Twitter的Streaming API出发，经过Tweepy的过滤规则筛选，进入队列缓冲区，然后分流到实时分析和持久化存储两个方向。同时，断线重连机制像安全网一样保护着整个流程。

我的建议： 刚开始做的时候，别追求完美。先让一条推文能完整地走通整个流程——从接收到分析到存储。然后再考虑优化性能、处理异常。我见过太多人一开始就想搞分布式、搞高可用，结果连最基本的流都没跑通。

好了，这一节的内容就到这。记住：实时数据流接入是整个情绪监控系统的「水源」。水源不干净、不稳定，后面的分析再牛也没用。把Tweepy的Streaming Client玩熟，你的系统就成功了一半。

专注资料整理
← 上一章 📖 返回目录 下一章 →