PySpark新手求助:基于Col1对比两文件并新增匹配状态列
PySpark实现基于Col1的匹配标记(新增Newcol列)
针对你的需求,这里提供两种适配不同数据量场景的高效实现方式:
方法1:使用isin判断(适合File2数据量较小的场景)
直接提取File2中所有唯一的Col1值,对File1每行进行匹配判断:
from pyspark.sql.functions import col, when # 读取文件(根据实际格式调整,比如parquet、json等) file1_df = spark.read.csv("path/to/your/file1.csv", header=True, inferSchema=True) file2_df = spark.read.csv("path/to/your/file2.csv", header=True, inferSchema=True) # 获取File2中不重复的Col1值集合 file2_unique_col1 = file2_df.select("Col1").distinct().rdd.flatMap(lambda x: x).collect() # 新增Newcol列:匹配填1,不匹配填0 result_df = file1_df.withColumn( "Newcol", when(col("Col1").isin(file2_unique_col1), 1).otherwise(0) ) # 查看结果 result_df.show()
方法2:使用广播+左连接(适合File2数据量大的场景)
当File2数据量较大时,collect()会将数据拉到Driver端引发内存压力,此时用广播变量+左连接更高效:
from pyspark.sql.functions import broadcast, when, col # 读取数据 file1_df = spark.read.csv("path/to/your/file1.csv", header=True, inferSchema=True) file2_df = spark.read.csv("path/to/your/file2.csv", header=True, inferSchema=True) # 提取File2的Col1并去重,广播这个小数据集优化连接性能 file2_col1_df = broadcast(file2_df.select("Col1").distinct()) # 左连接后判断匹配状态,生成Newcol并移除重复列 result_df = file1_df.join(file2_col1_df, on="Col1", how="left") \ .withColumn("Newcol", when(col(file2_col1_df.Col1).isNotNull(), 1).otherwise(0)) \ .drop(file2_col1_df.Col1) # 查看结果 result_df.show()
结果验证
针对你提供的示例数据,两种方法都会生成预期输出:
| Col1 | Col2 | ... | ColN | Newcol |
|---|---|---|---|---|
| 1 | abc | ... | Xxxx | 1 |
| 2 | abc | ... | Xxxx | 1 |
| 3 | abc | ... | Xxxx | 0 |
内容的提问来源于stack exchange,提问作者Naji
相关产品推荐
相关产品推荐

