Spark中如何获取两个RDD的交集并提取对应数据？

针对你的需求，核心是从RDD1中筛选出col1字段存在于RDD2中的完整记录，以下是两种高效实现方法：

方法一：广播变量过滤（适合RDD2数据量较小的场景） #

通过广播RDD2的有效键集合，在Executor端直接过滤RDD1，避免大量数据传输：

1. 处理RDD2，提取并清理col1的值，收集为本地集合后广播到所有节点
2. 遍历RDD1，保留col1在广播集合中的记录

Python代码示例 #

from pyspark import SparkContext

sc = SparkContext("local", "RDDIntersection")

# 初始化RDD数据（如果是从文件读取，需先处理表头）
rdd1 = sc.parallelize([("A", "x123"), ("B", "y123"), ("C", "z123")])
# 清理RDD2中的空格和空行
rdd2 = sc.parallelize(["A", "C"]).map(lambda x: x.strip())

# 广播有效键集合
valid_col1 = set(rdd2.collect())
broadcast_col1 = sc.broadcast(valid_col1)

# 过滤得到目标RDD3
rdd3 = rdd1.filter(lambda item: item[0] in broadcast_col1.value)

# 输出结果
for row in rdd3.collect():
    print(f"{row[0]}     {row[1]}")

方法二：内连接操作（适合RDD2数据量较大的场景） #

通过分布式内连接，避免将大量数据拉取到Driver端，更适合大数据场景：

1. 将RDD2转换为键值对RDD（键为col1，值可设为占位符）
2. 与RDD1执行内连接，提取RDD1的原始记录

Python代码示例 #

from pyspark import SparkContext

sc = SparkContext("local", "RDDJoinExample")

rdd1 = sc.parallelize([("A", "x123"), ("B", "y123"), ("C", "z123")])
# RDD2转成键值对格式
rdd2_kv = sc.parallelize(["A", "C"]).map(lambda x: (x.strip(), None))

# 内连接后提取原始数据
rdd3 = rdd1.join(rdd2_kv).map(lambda x: (x[0], x[1][0]))

# 输出结果
for row in rdd3.collect():
    print(f"{row[0]}     {row[1]}")

处理带表头的原始数据 #

如果你的RDD是从文件读取的带表头数据，需先跳过表头行：

# 处理RDD1（带表头）
rdd1_raw = sc.textFile("path/to/rdd1.txt")
header1 = rdd1_raw.first()
rdd1_data = rdd1_raw.filter(lambda line: line != header1).map(lambda line: line.split())

# 处理RDD2（带表头）
rdd2_raw = sc.textFile("path/to/rdd2.txt")
header2 = rdd2_raw.first()
rdd2_data = rdd2_raw.filter(lambda line: line != header2).map(lambda line: line.strip())

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