我想将Dstream
转换为DataFrame
,以便对这个DataFrame
应用相同的转换,并调用NaiveBayesModel
模型来预测目标概率,我使用apachespark 2.1.1,Dstream
是从socketTextStream
构建的。我试图调用foreachRDD
函数的foreachRDD
,但它没有工作。在
def predict(rdd):
count = rdd.count()
if(count>0):
hashingTF = HashingTF(numFeatures=1000)
features = hashingTF.transform(rdd)
result = model.transform(features)
return result.probability
else:
print("No data receveid")
model = NaiveBayesModel.load(sc, "ML_models/NaiveClassifier/naiveBayesClassifier-2010-09-10-08-51-25")
lines = ssc.socketTextStream("localhost", 9999)
tweets = lines.map(lambda v: json.loads(v))
text_dstream = tweets.map(lambda tweet: tweet['text'])
df = text_dstream.foreachRDD(lambda rdd: predict(rdd))
ssc.start() # Start the computation
ssc.awaitTermination()
我收到以下错误消息
^{pr2}$我的想法是将Dstream
转换为SparkDataFrame
,并使用以下方法应用转换:
#Tokenize sentiment text
tokenizer = Tokenizer(inputCol="SentimentText", outputCol="SetimentTextTokenize")
wordsData = tokenizer.transform(df)
hashingTF = HashingTF(inputCol="SetimentTextTokenize", outputCol="rawFeatures", numFeatures=1000)
featurizedData = hashingTF.transform(wordsData)
目前没有回答
相关问题 更多 >
编程相关推荐