PySpark 地图不工作答案

【问题标题】：PySpark map not workingPySpark 地图不工作
【发布时间】：2016-07-30 23:16:32
【问题描述】：

我是 Apache Spark 的新手，一个简单的地图函数实现为

from pyspark import  SparkContext
sc = SparkContext( 'local', 'pyspark')

f = open("Tweets_tokenised.txt")
tokenised_tweets = f.readlines()

f = open("positive.txt")
pos_words=f.readlines()
f = open("negative.txt")
neg_words=f.readlines()
def sentiment(line):
    global pos_words
    global neg_words
    pos = 0
    neg = 0

    for word in line.split():
        if word in pos_words:
            pos=pos+1

        if word in neg_words:
            neg=neg+1

    if(pos > neg):
        return 1
    else:
        return 0
dist_tweets=sc.textFile("Tweets_tokenised.txt").map(sentiment)
#(lambda line: sentiment(line))
dist_tweets.saveAsTextFile("RDD.txt")

基本上，我正在读取一个文件（包含标记化和词干化的推文），然后在 map 函数中对其进行简单的正负字数统计。（最后的第 3 行）但是 RDD.txt 里面什么都没有。根本没有调用函数情感。谁能指出错误

【问题讨论】：

标签： apache-spark mapreduce pyspark

【解决方案1】：

您无法更改Apache Spark 中的map 转换中的全局变量的值，您需要一个Accumulator，但即使使用它们，我认为这不是正确的方法。

如果您的pos_words 和neg_words 不是那么大，您可以将它们定义为Broadcast 列表，然后按sentiment 计数。

类似：

pos = sc.broadcast(["good", "gold", "silver"])
neg = sc.broadcast(["evil", "currency", "fiat"])

# I will suppose that every record is a different tweet and are stored in tuples.
tweets = sc.parallelize([("banking", "is", "evil"), ("gold", "is", "good")])

(tweets
 .flatMap(lambda x: x)
 .map(lambda x: (1 if x in pos.value else -1 if x in neg.value else 0, 1))
 .reduceByKey(lambda a, b: a + b).take(3))

# notice that I count neutral words.
# output -> [(0, 3), (1, 2), (-1, 1)]

注意，你可以在here查看示例。

PD：如果您的想法是计算每条消息的正面和负面词，那么方法会略有不同。

【讨论】：