SPARK STREAMING (PYTHON) 中的 % 字数

Question

在下一个示例中，我从 Kafka 接收到一个序列词：

('cat')
('dog')
('rat')
('dog')

我的目标是计算每个单词的历史百分比。我将有两个 RDD，一个带有历史字数，另一个带有所有单词的总数：

values = KafkaUtils.createDirectStream(ssc, [topic], {"metadata.broker.list": brokers})


def updatefunc (new_value, last_value):
    if last_value is None:
        last_value = 0
    return sum(new_value, last_value)


words=values.map(lambda x: (x,1)).reduceByKey(lambda a,b: a+b)

historic= words.updateStateByKey(updatefunc).\
    transform(lambda  rdd: rdd.sortBy(lambda (x,v): x))

totalNo = words.\
    map(lambda x: x[1]).reduce(lambda a,b:a+b).map(lambda x: (('totalsum',x))).updateStateByKey(updatefunc).map(lambda x:x[1])

现在我正在尝试除以：((每个键的历史值)/totalNo)*100 来获得每个单词的百分比：

solution=historic.map(lambda x: x[0],x[1]*100/totalNo)

但我得到了错误：

 It appears that you are attempting to reference SparkContext from a broadcast variable, action, or transforamtion. SparkContext can only be used on the driver, not in code that it run on workers. For more information, see SPARK-5063

如何修复totalNO 的值，以便在另一个RDD 中使用它？

Answer 1

最后这个方法也可以了：

words = KafkaUtils.createDirectStream(ssc, topics=['test'], kafkaParams={'bootstrap.servers': 'localhost:9092'})\
    .map(lambda x: x[1]).flatMap(lambda x: list(x))

historic = words.map(lambda x: (x, 1)).updateStateByKey(lambda x, y: sum(x) + (y or 0))

def func(rdd):
    if not rdd.isEmpty():
        totalNo = rdd.map(lambda x: x[1]).reduce(add)
        rdd = rdd.map(lambda x: (x[0], x[1] / totalNo))
    return rdd

solution = historic.transform(func)

solution.pprint()

这是你想要的吗？