PySpark新手求助：基于Col1对比两文件并新增匹配状态列

针对你的需求，这里提供两种适配不同数据量场景的高效实现方式：

方法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()

结果验证 #

针对你提供的示例数据，两种方法都会生成预期输出：

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