基于Spark与Kafka的Twitter流数据：如何存储至MongoDB

Hey，我来帮你把这套Twitter流采集到Spark Streaming再存储到MongoDB的流程整理清楚，结合你提到的代码片段给你整合出完整可参考的实现：

1. Twitter流采集核心代码 #

这部分负责抓取Twitter实时流数据并发送到Kafka主题，是整个流程的数据源入口：

import tweepy
from kafka import KafkaProducer
import json

# 替换成你自己的Twitter API密钥
consumer_key = "YOUR_CONSUMER_KEY"
consumer_secret = "YOUR_CONSUMER_SECRET"
access_token = "YOUR_ACCESS_TOKEN"
access_token_secret = "YOUR_ACCESS_TOKEN_SECRET"

# 初始化Kafka生产者，指定Broker地址
producer = KafkaProducer(bootstrap_servers='localhost:9092',
                         value_serializer=lambda v: json.dumps(v).encode('utf-8'))
target_topic = 'topic1'

class TwitterStreamHandler(tweepy.StreamListener):
    def on_data(self, data):
        # 将Twitter原始数据发送到Kafka主题
        producer.send(target_topic, data)
        print(f"Sent tweet to Kafka: {data[:50]}...")  # 打印部分数据做验证
        return True

    def on_error(self, status):
        print(f"Twitter Stream Error: {status}")

if __name__ == "__main__":
    auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
    auth.set_access_token(access_token, access_token_secret)
    # 启动Twitter流，可根据需求添加关键词过滤
    stream = tweepy.Stream(auth, TwitterStreamHandler())
    stream.filter(track=['spark', 'kafka'], languages=['en'])

2. Spark Streaming处理并写入MongoDB #

这部分从Kafka消费数据，完成流处理后将数据持久化到MongoDB：

from pyspark import SparkConf, SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils
import json
from pymongo import MongoClient

def batch_write_to_mongo(rdd):
    """自定义函数：将每个RDD批次的数据写入MongoDB"""
    if not rdd.isEmpty():
        # 连接MongoDB，替换成你的数据库地址和库名
        client = MongoClient('mongodb://localhost:27017/')
        db = client['twitter_stream_db']
        tweet_collection = db['raw_tweets']
        
        # 解析每条JSON格式的Twitter数据并插入
        for tweet_json in rdd.collect():
            tweet_data = json.loads(tweet_json)
            tweet_collection.insert_one(tweet_data)
        
        client.close()
        print(f"Inserted {rdd.count()} tweets to MongoDB")

def main():
    conf = SparkConf().setMaster("local[2]").setAppName("TwitterStreamProcessor")
    sc = SparkContext(conf=conf)
    sc.setLogLevel("WARN")  # 减少冗余日志输出
    # 每10秒生成一个处理批次
    ssc = StreamingContext(sc, 10)
    ssc.checkpoint("checkpoint")  # 设置检查点，保证流处理容错

    # Kafka消费者配置
    kafka_params = {
        "metadata.broker.list": "localhost:9092",
        "auto.offset.reset": "smallest"  # 从最早的偏移量开始消费
    }
    target_topics = ['topic1']

    # 创建Kafka直接流，获取消息的value部分（Twitter数据）
    kafka_stream = KafkaUtils.createDirectStream(ssc, target_topics, kafka_params)
    tweet_stream = kafka_stream.map(lambda msg_tuple: msg_tuple[1])

    # 对每个批次的RDD执行写入MongoDB操作
    tweet_stream.foreachRDD(batch_write_to_mongo)

    # 启动流处理并等待终止
    ssc.start()
    ssc.awaitTermination()

if __name__ == "__main__":
    main()

关键注意事项 #

• 提前安装依赖：pip install tweepy kafka-python pymongo pyspark
• 确保Kafka、MongoDB服务正常启动，端口（默认9092、27017）可访问
• Twitter API密钥需要在Twitter开发者平台申请并替换
• 生产环境中，setMaster("local[2]")要替换成Spark集群的Master地址

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