您的输入字符串:
# spark is SparkSession object
s1 = 'The Adventure of the Blue Carbuncle The Adventure of the Blue Carbuncle The Adventure of the Blue Carbuncle; The Adventure of the Blue Carbuncle;'
# Split the string on ; and I parallelize it to make an rdd
rddData = spark.sparkContext.parallelize(rdd_Data.split(";"))
rddData.collect()
# ['The Adventure of the Blue Carbuncle The Adventure of the Blue Carbuncle The Adventure of the Blue Carbuncle', ' The Adventure of the Blue Carbuncle', '']
import itertools
final = (
rddData.filter(lambda x: x != "")
.map(lambda x: x.split(" "))
.flatMap(lambda x: itertools.combinations(x, 2))
.filter(lambda x: x[0] != "")
.map(lambda x: (x, 1))
.reduceByKey(lambda x, y: x + y).collect()
)
# [(('The', 'of'), 7), (('The', 'Blue'), 7), (('The', 'Carbuncle'), 7), (('Adventure', 'the'), 7), (('Adventure', 'Adventure'), 3), (('of', 'The'), 3), (('the', 'Adventure'), 3), (('the', 'the'), 3), (('Blue', 'The'), 3), (('Carbuncle', 'The'), 3), (('Adventure', 'The'), 3), (('of', 'the'), 7), (('of', 'Adventure'), 3), (('the', 'The'), 3), (('Blue', 'Adventure'), 3), (('Blue', 'the'), 3), (('Carbuncle', 'Adventure'), 3), (('Carbuncle', 'the'), 3), (('The', 'The'), 3), (('of', 'Blue'), 7), (('of', 'Carbuncle'), 7), (('of', 'of'), 3), (('Blue', 'Carbuncle'), 7), (('Blue', 'of'), 3), (('Blue', 'Blue'), 3), (('Carbuncle', 'of'), 3), (('Carbuncle', 'Blue'), 3), (('Carbuncle', 'Carbuncle'), 3), (('The', 'Adventure'), 7), (('The', 'the'), 7), (('Adventure', 'of'), 7), (('Adventure', 'Blue'), 7), (('Adventure', 'Carbuncle'), 7), (('the', 'Blue'), 7), (('the', 'Carbuncle'), 7), (('the', 'of'), 3)]
- 删除第一个分割中的所有空格
- 将 x 分割为空格分隔的字符串,按空格
- 创建 2 个元素的组合,每个元素使用
itertools.combinations
(flatMap
将每个单词与行中的每个其他单词配对)
- 像字数统计一样进行映射和缩减