pyspark: как использовать оператор when в pyspark с двумя фреймами данных

Question

У меня есть три фрейма данных pyspark:

1. транзакция (column_names = ['result_3', 'ID')
2. rule_3 (column_names = ['OrderID']) *
3. rule_4 (column_names = ['OrderID'])

Таблица транзакций выглядит примерно так:

result_3|result_4|ID
0       |0       |11
0       |0       |13 
0       |0       |10 
0       |0       |9 
0       |0       |7

Таблица rule_3 имеет вид:

OrderID
13 
9 

Таблица rule_4 гласит:

OrderID
9 
7 

Если в файлеaction.ID также существует элемент правила rule_3.OrderID, я бы хотел установить для 1. сопровождающего элемента столбца транзакции.result_3 значение 1. То же применяется аналогично таблице result_4.

Таким образом, результат должен выглядеть следующим образом:

result_3|result_4|ID
0       |0       |11
1       |0       |13 
0       |0       |10 
1       |1       |9 
0       |1       |7

Если пробовали эту строку кода (только относительно result_3), но она не работает

test = transaction.withColumn("result_3", F.when(F.col("ID").isin(rule_3.OrderID), 1).otherwise(0))

Сообщение об ошибке:

pyspark.sql.utils.AnalysisException: 
'Resolved attribute(s) OrderID#8 missing from 6#21L,5#20L,8#22L,4#19L,3#18L,ID#23 in operator !Project [3#18L, 4#19L, 5#20L, 6#21L, 8#22L, CASE WHEN ID#23 IN (OrderID#8) THEN 1 ELSE 0 END AS ID#131].;;
!Project [3#18L, 4#19L, 5#20L, 6#21L, 8#22L, CASE WHEN ID#23 IN (OrderID#8) THEN 1 ELSE 0 END AS ID#131]
+- LogicalRDD [3#18L, 4#19L, 5#20L, 6#21L, 8#22L, ID#23], false'

Спасибо!

ОБНОВЛЕНИЕ

Было бы также неплохо сделать то же самое в динамический способ (пока не работает):

rules = [rule_3, rule_4]
result = transaction.select(F.col("ID").alias("OrderID")
for i in range(2):
   result = result.join(
    rules[i].select(
        "OrderID",
        F.lit(1).alias(str(rules[i])),
    ),
    on="OrderID",
    how="left",
                    )

Сообщение об ошибке гласит: {ValueError} Истинное значение Series является неоднозначным uous. Используйте a.empty, a.bool (), a.item (), a.any () или a.all ().

Answer 1

вы можете сделать это с помощью простого соединения.

from pyspark.sql import functions as F

df = transaction.select(
    F.col("ID").alias("OrderID")
).join(
    rule_3.select(
        "OrderID", 
        F.lit(1).alias("result_3"),
    ),
    on="OrderID",
    how="left",
).join(
    rule_4.select(
        "OrderID", 
        F.lit(1).alias("result_4"),
    ),
    on="OrderID",
    how="left",
)

df.select(
    F.coaslesce(
        F.col("result_3"), 
        F.lit(0)
    ).alias("result_3"),
    F.coaslesce(
        F.col("result_4"), 
        F.lit(0)
    ).alias("result_4"),
    F.col("OrderID").alias("ID"),
).show()