PySpark - Check from a list of values are present in any of the columns in a Dataframe

user_2210672 0
Viewed 147

I have a Dataframe 'DF' with 200 columns and around 500 million records. I have another list of values as 'l'. I want to check, for each of the values in the list l, each of the value is present in which column in my Dataframe DF.

For example,

l = [101,102,103,104,105]

DF =

col1,col2,col3

1,2,101

2,3,102

3,103,4

4,104,5

105,5,6

6,105,7

My output should be as below:

101,col3

102,col3

103,col2

104,col2

105,col1

105,col2
pyspark apache-spark-sql apache-spark python
2 Answers

Unpivot the dataframe using stack, and then filter the dataframe using isin:

import pyspark.sql.functions as F

l = [101,102,103,104,105]

df2 = df.selectExpr("""
    stack(3, col1, 'col1', col2, 'col2', col3, 'col3') as (val, col)
""").filter(
    F.col('val').isin(l)
)

df2.show()
+---+----+
|val| col|
+---+----+
|101|col3|
|102|col3|
|103|col2|
|104|col2|
|105|col1|
|105|col2|
+---+----+
user_141657300

Another way without unpivoting the data frame :

from pyspark.sql import functions as F

df1 = df.withColumn(
    "presence_arr",
    F.array(*[
        F.when(F.col(c).isin(l), F.struct(F.lit(c).alias("col"), F.col(c).alias("val")))
        for c in df.columns
    ])) \
    .withColumn("presence_arr", F.expr(f"filter(presence_arr, x -> x is not null)")) \
    .agg(F.collect_set("presence_arr").alias("presence_arr")) \
    .withColumn("presence_arr", F.explode(F.flatten("presence_arr"))) \
    .select("presence_arr.*")

df1.show(truncate=False)

#+----+---+
#|col |val|
#+----+---+
#|col2|105|
#|col3|101|
#|col3|102|
#|col1|105|
#|col2|104|
#|col2|103|
#+----+---+

For each row, we check each column if it's present in the list of values, then agg to collect all the arrays, flatten and explode to get the desired output.

user_13865510
Related