嘿，这个需求其实挺常见的，你想到用配对的思路完全没问题！本质上就是要把文本里的每个单词和它紧挨着的下一个单词组成键值对，下面我给你两种实现方式，还会帮你分析哪种更优～

方法一：分区内本地生成配对（推荐！无Shuffle，性能更高） #

因为Spark的RDD每个分区是本地有序的列表，我们可以直接在每个分区里处理相邻单词的配对，不需要跨节点交换数据，这是最高效的方式。

以Python为例：

# 第一步：读取文本文件，分割成单个单词的RDD
# 这里可以根据需求做清洗，比如转小写、过滤标点
words_rdd = sc.textFile("your_text_file.txt").flatMap(lambda line: line.split())

# 定义一个分区处理函数，生成相邻单词对
def generate_word_pairs(partition_iter):
    # 把分区迭代器转成列表，方便遍历
    word_list = list(partition_iter)
    # 遍历到倒数第二个单词，和下一个单词配对
    for idx in range(len(word_list) - 1):
        yield (word_list[idx], word_list[idx + 1])

# 生成最终的Pair RDD
pair_rdd = words_rdd.mapPartitions(generate_word_pairs)

如果是Scala的话，代码会更简洁：

val wordsRDD = sc.textFile("your_text_file.txt").flatMap(_.split(" "))

def generateWordPairs(iter: Iterator[String]): Iterator[(String, String)] = {
    val wordList = iter.toList
    wordList.zip(wordList.tail).iterator
}

val pairRDD = wordsRDD.mapPartitions(generateWordPairs)

方法二：用索引关联（有Shuffle，适合小数据量） #

如果你坚持想用类似zip的思路，也可以通过给单词加索引的方式实现，但这种方法会触发Shuffle（因为要按索引关联），大数据量下性能不如第一种方法：

Python示例：

# 给每个单词加上索引
indexed_words = words_rdd.zipWithIndex()
# 把下一个单词的索引减1，方便和前一个单词关联
next_word_rdd = indexed_words.map(lambda x: (x[1] - 1, x[0]))
# 按索引join，然后提取键值对
pair_rdd = indexed_words.join(next_word_rdd).map(lambda x: (x[1][0], x[1][1]))

额外小提示 #

如果需要处理标点、大小写这类细节，可以在分割单词的时候做清洗，比如用正则提取有效单词：

import re
words_rdd = sc.textFile("your_text_file.txt").flatMap(lambda line: re.findall(r'\w+', line.lower()))

这样就能把"Hello!"变成"hello"，避免标点导致的单词不匹配问题～

函数式编程的思路其实就是把复杂的任务拆成一个个小的、纯函数的操作，交给框架去分布式执行，多写几次就慢慢习惯啦！

内容的提问来源于stack exchange，提问作者Bassinator