在 Python 中使用 kafka 生产者发送数据时出现问题（Jupyter Notebook）

Question

我正在尝试使用 Kafka、Python 和 Twitter 创建大数据分析。我有一个推文数据流，我只使用它们的标签。我的问题与 Kafka 用于 Python 的生产者有关。我无法将我想要的数据发送到我创建的主题中，因为我没有看到任何选项可以向生产者发送变量的内容。

在https://kafka-python.readthedocs.io/en/master/usage.html 中，我只能看到使用b'some_string' 发送精确字符串的选项。但我想发送我从 Twitter 流中获取的主题标签。我对 Python 了解不多，如果解决方案很明显，请见谅。

进口：

from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils
import json
import tweepy
from tweepy import OAuthHandler
from tweepy import Stream
import kafka
from kafka import SimpleProducer, KafkaClient
from kafka import KafkaProducer

流媒体上下文：

ssc = StreamingContext(sc,60)

键：

consumer_key="consumer_key"
consumer_secret="consumer_secret"
access_token="access_token"
access_token_secret="access_token_secret"

Tweepy：

auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
auth.set_access_token(access_token, access_token_secret)
api = tweepy.API(auth)

制作人：

producer = KafkaProducer(bootstrap_servers=['localhost:9092'])

代码：

class MyStreamListener(tweepy.StreamListener):

    def on_status(self, status):
        for hashtag in status.entities['hashtags']:
            prueba = b'hashtag["text"]'
            producer.send('topic', prueba)
            return True
    def on_error(self, status_code):
        if status_code == 420:
            #returning False in on_data disconnects the stream
            return False

流监听器：

myStreamListener = MyStreamListener()
myStream = tweepy.Stream(auth = api.auth, listener=MyStreamListener())

推文流：

myStream.filter(track=['some_text'])

问题是，生产者只发送prueba的文字字符串，即"(hashtag["text"])"。我想发送的不是确切的东西，而是它的内容。

提前致谢。

Answer 1

producer.send('topic', hashtag) 怎么样？您还需要确保将数据编码为原始字节，这是 kafka 存储的内容。如果主题标签是一个简单的字符串，您可以使用 producer.send('topic', hashtag.encode('utf-8')) 。如果是 dict 或更复杂的数据结构，可能需要在编码为字节之前使用 json.dumps。希望这会有所帮助！