我试图在cloudera vm上使用pyspark和hive创建数据框,但每次我都会收到此错误 .
回溯(最近一次调用最后一次):文件“/home/cloudera/Desktop/TwitterSentimentAnalysis/SentimentAnalysis.py”,第98行,在.reduceByKey(lambda a,b:ab)\ File“/ usr / lib / spark / python / lib / pyspark.zip / pyspark / sql / context.py“,第62行,在toDF文件中”/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/context.py“,404行, createDataFrame文件“/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/context.py”,第285行,在_createFromRDD文件中“/usr/lib/spark/python/lib/pyspark.zip/pyspark /sql/context.py“,第229行,在_inferSchema文件”/usr/lib/spark/python/lib/pyspark.zip/pyspark/rdd.py“,第1320行,第一个ValueError:RDD为空
INFO spark.SparkContext:从关闭钩子调用stop()
我该怎么做才能解决这个错误 .
编辑2 - sc = SparkContext(appName =“PythonSentimentAnalysis”)sqlCtx = HiveContext(sc)
filenameAFINN = "/home/cloudera/Desktop/TwitterSentimentAnalysis/AFINN/AFINN-111.txt"
afinn = dict(map(lambda (w, s): (w, int(s)), [ ws.strip().split('\t') for ws in open(filenameAFINN) ]))
filenameCandidate = "file:///home/cloudera/Desktop/TwitterSentimentAnalysis/Candidates/Candidate Mapping.txt"
candidates = sc.textFile(filenameCandidate).map(lambda x: (x.strip().split(",")[0],x.strip().split(","))) \
.flatMapValues(lambda x:x).map(lambda y: (y[1],y[0])).distinct()
pattern_split = re.compile(r"\W+")
tweets = sqlCtx.sql("select id, text, entities.user_mentions.name from incremental_tweets")
def sentiment(text):
words = pattern_split.split(text.lower())
sentiments = map(lambda word: afinn.get(word, 0), words)
if sentiments:
sentiment = float(sum(sentiments))/math.sqrt(len(sentiments))
else:
sentiment = 0
return sentiment
sentimentTuple = tweets.rdd.map(lambda r: [r.id, r.text, r.name]) \
.map(lambda r: [sentiment(r[1]),r[2]]) \
.flatMapValues(lambda x: x) \
.map(lambda y: (y[1],y[0])) \
.reduceByKey(lambda x, y: x+y) \
.sortByKey(ascending=True)
scoreDF = sentimentTuple.join(candidates) \
.map(lambda (x,y): (y[1],y[0])) \
.reduceByKey(lambda a,b: a+b) \
.toDF()
scoreRenameDF = scoreDF.withColumnRenamed("_1","Candidate").withColumnRenamed("_2","Score")
sqlCtx.registerDataFrameAsTable(scoreRenameDF, "SCORE_TEMP")
sqlCtx.sql("INSERT OVERWRITE TABLE candidate_score SELECT Candidate, Score FROM SCORE_TEMP")
1 回答
如果正确创建,请尝试使用下面的代码检查中间RDD:
这将显示您的RDD的前10个条目