第一关 WordCount - 词频统计
- 对文本文件内的每个单词都统计出其出现的次数。
- 按照每个单词出现次数的数量,降序排序。
文本文件内容如下(单词与单词之间以空格进行分割):
hello java
hello python java
hello python python
hello flink
scala scala scala scala scala
预期输出: (scala,5)
(hello,4)
(python,3)
(java,2)
(flink,1)
# -*- coding: UTF-8 -*-
from pyspark import SparkContext
if __name__ == "__main__":
"""
需求:对本地文件系统URI为:/root/wordcount.txt 的内容进行词频统计
"""
# ********** Begin **********#
sc = SparkContext("local","pySpark")
rdd = sc.textFile("/root/wordcount.txt")
values = rdd.flatMap(lambda x:str(x).split(" ")).map(lambda x:(x,1)).reduceByKey(lambda x,y:x+y).sortBy(lambda x:tuple(x)[1],False)
print(values.collect())
# ********** End **********#
第二关 Friend Recommendation - 好友推荐
统计间接好友并输出结果。
预期结果:[(world_tom,2),(tom_mr,1),(mr_hadoop,1),(cat_mr,1),(world_cat,1),(cat_hadoop,2),(hive_tom,3),(world_mr,2)
]
# -*- coding: UTF-8 -*-
from pyspark import SparkContext
def word_couple(word1, word2):
if hash(word1) > hash(word2):
return word1 + '_' + word2
return word2 + '_' + word1
def relations(items):
result = []
for i in range(1, len(items)):
result.append((word_couple(items[0], items[i]), 0))
for j in range(i+1, len(items)):
result.append((word_couple(items[i], items[j]), 1))
return result
def fun2(x):
values = tuple(x[1])
return ((x[0], 0) if min(values)==0 else (x[0], sum(values)))
if __name__ == "__main__":
"""
需求:对本地文件系统URI为:/root/friend.txt 的数据统计间接好友的数量
"""
# ********** Begin **********#
sc = SparkContext("local", "friend recommendation")
src = sc.textFile("/root/friend.txt").map(lambda x:x.strip().encode('utf-8').split(" "))
rdd = src.flatMap(relations).reduceByKey(lambda x,y:0 if x==0 or y==0 else x+y).filter(lambda x:x[1]>0)
print(rdd.collect())
# ********** End **********#
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)