如果某列在另一个Spark数据表中，Pyspark则根据该列创建新列

Question

我试图在Spark Dataframe中创建一个列，如果列的行位于单独的Dataframe中，则为标志。

这是我的主Spark Dataframe(df_main)

+--------+
|main    |
+--------+
|28asA017|
|03G12331|
|1567L044|
|02TGasd8|
|1asd3436|
|A1234567|
|B1234567|
+--------+

这是我的引用(df_ref)，这个引用中有数百行，所以我显然不能像这个解决方案或这个解决方案那样硬编码它们

+--------+
|mask_vl |
+--------+
|A1234567|
|B1234567|
...
+--------+

df_main['is_inref'] = np.where(df_main['main'].isin(df_ref.mask_vl.values), "YES", "NO")

+--------+--------+
|main |is_inref|
+--------+--------+
|28asA017|NO      |
|03G12331|NO      |
|1567L044|NO      |
|02TGasd8|NO      |
|1asd3436|NO      |
|A1234567|YES     |
|B1234567|YES     |
+--------+--------+

我已经尝试了下面的代码，但我不明白图片中的错误是什么意思。

df_main = df_main.withColumn('is_inref', "YES" if F.col('main').isin(df_ref) else "NO")
df_main.show(20, False)

Answer 1

你已经接近了。我认为您需要的附加步骤是显式创建包含df_ref中的值的列表。

请看下面的插图：

# Create your DataFrames
df = spark.createDataFrame(["28asA017","03G12331","1567L044",'02TGasd8','1asd3436','A1234567','B1234567'], "string").toDF("main")
df_ref =  spark.createDataFrame(["A1234567","B1234567"], "string").toDF("mask_vl")

然后，您可以创建一个列表并使用isin，几乎与您所使用的一样：

# Imports
from pyspark.sql.functions import col, when

# Create a list with the values of your reference DF
mask_vl_list = df_ref.select("mask_vl").rdd.flatMap(lambda x: x).collect()

# Use isin to check whether the values in your column exist in the list
df_main = df_main.withColumn('is_inref', when(col('main').isin(mask_vl_list), 'YES').otherwise('NO'))

>>> df_main.show()

+--------+--------+
|    main|is_inref|
+--------+--------+
|28asA017|      NO|
|03G12331|      NO|
|1567L044|      NO|
|02TGasd8|      NO|
|1asd3436|      NO|
|A1234567|     YES|
|B1234567|     YES|
+--------+--------+