Вы можете достичь этого, используя SQL выражений, которые можно оценить, используя expr. Однако вы не сможете присоединиться к двум фреймам данных, поскольку выражения SQL не могут быть оценены как значения столбцов (см. post ), поэтому необходимо собрать функции в список (как у вас есть только сотни строк, это может уместиться в памяти).
Вот рабочий пример, который вы можете адаптировать под свои требования:
data1 = [(1, "val1", 4, 5, "A", 10), (0, "val2", 7, 8, "B", 20),
(9, "val3", 8, 1, "C", 30), (10, "val4", 2, 9, "D", 30),
(20, "val5", 6, 5, "E", 50), (3, "val6", 100, 2, "X", 45)]
df1 = spark.createDataFrame(data1, ["col1", "col2", "col3", "col4", "col5", "col6"])
data2 = [("func1", "col1 + col3 = 5 and col2 like '%al1'"),
("func2", "col6 = 30 or col1 * col4 > 20"),
("func3", "col5 in ('A', 'B', 'C') and col6 - col1 < 30"),
("func4", "col2 like 'val%' and col1 > 0")]
df2 = spark.createDataFrame(data2, ["func_name", "func_body"])
# get functions into a list
functions = df2.collect()
# case/when expression to evaluate the functions
satisfied_expr = [when(expr(f.func_body), lit(f.func_name)) for f in functions]
# add new column tags
df1.withColumn("tags", array(*satisfied_expr)) \
.withColumn("tags", expr("filter(tags, x -> x is not null)")) \
.show(truncate=False)
После добавления столбца массива tags, filter используется функция для удаления нулевые значения, которые соответствуют неудовлетворенным выражениям. Эта функция доступна только начиная с Spark 2.4+, вам придется использовать и UDF для более старых версий.
Дает:
+----+----+----+----+----+----+---------------------+
|col1|col2|col3|col4|col5|col6|tags |
+----+----+----+----+----+----+---------------------+
|1 |val1|4 |5 |A |10 |[func1, func3, func4]|
|0 |val2|7 |8 |B |20 |[func3] |
|9 |val3|8 |1 |C |30 |[func2, func3, func4]|
|10 |val4|2 |9 |D |30 |[func2, func4] |
|20 |val5|6 |5 |E |50 |[func2, func4] |
|3 |val6|100 |2 |X |45 |[func4] |
+----+----+----+----+----+----+---------------------+