如何将嵌套SQL NOT IN查询转换为PySpark函数式代码？

环境信息：

• Spark version: 2.3
• Hive: Hive 1.1.0-cdh5.16.2
• Engine: PySpark (Spark SQL)

需要将以下SQL语句改写为PySpark函数式查询：

select distinct id_p
from table1
where tab1.id_p not in (select distinct id_p from table2)

已有部分PySpark代码，需补全isin方法的参数以实现等价逻辑：

tab1 = spark.table("table1")
tab2 = spark.table("table2")
tab1 = tab1.select("id_p").distinct()
tab1 = tab1.filter(f.col("id_p").isin(??????) == False )

from pyspark.sql import functions as f

tab1 = spark.table("table1")
tab2 = spark.table("table2")

# 提取table2中去重后的id_p并转为本地列表
tab2_distinct_ids = tab2.select("id_p").distinct().rdd.flatMap(lambda x: x).collect()

tab1 = tab1.select("id_p").distinct()
# 补全isin参数，通过取反实现not in逻辑
tab1 = tab1.filter(f.col("id_p").isin(tab2_distinct_ids) == False)

1. 原SQL的not in子查询逻辑，对应PySpark中先获取子查询结果集，再用isin配合取反实现。
2. Spark 2.3版本中isin无法直接接收DataFrame作为参数，必须通过collect()将分布式数据拉取到Driver端转为本地列表。但如果table2的id_p数据量极大，collect()可能导致Driver内存溢出，这种场景更推荐用左外连接+过滤空值的方式实现，避免内存风险：

# 大数据量场景的替代方案
tab1_distinct = tab1.select("id_p").distinct()
tab2_distinct = tab2.select("id_p").distinct().withColumnRenamed("id_p", "tab2_id_p")

result = tab1_distinct.join(tab2_distinct, tab1_distinct.id_p == tab2_distinct.tab2_id_p, "left_outer")\
    .filter(f.col("tab2_id_p").isNull())\
    .select("id_p")

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