Spark check row exists in group using windowing

row_exists_in_group.py

from pyspark.sql.functions import *
from pyspark.sql.window import Window
from pyspark.sql import Column

from functools import reduce


w = Window.partitionBy('group_id')

filter_expression = reduce(Column.__and__, (
    array_contains(collect_list('value').over(w), val)
    for val in ('banana', 'orange')
))

display(
    df
    .where(col('group_id').isNotNull())
    .withColumn('has_all_fruits', filter_expression)
    .where('has_all_fruits')
    .drop('has_all_fruits')
)