Pyspark：在一个df中使用一行来过滤和选择另一个

Question

我正在尝试遍历一个 PySpark 数据帧中的行，并使用每一行中的值对第二个 Pyspark 数据帧执行操作（过滤、选择），然后绑定所有结果。也许这是最好的说明：

DF1

id   name   which_col
1    John   col1
2    Jane   col3
3    Bob    col2
4    Barb   col1

DF2

name  col1  col2  col3
Bob   78    43    54
Bob   23    65    34
Bob   12    98    75
John  45    54    54
John  75    43    12
Jane  24    45    21
...

我想对 DF1 中的每一行执行的步骤是：

1. 取“name”中的值并使用它来过滤 DF2（例如，对于第 1 行，将 DF2 过滤为仅“John”行。）
2. 然后根据“which_col”中的 DF1 值选择对应的 DF2 列（例如，对于 John，选择 DF2 中的 col1，而对于 Jane，则选择 col3）。
3. 对 DF1 的每一行重复
4. 将所有结果绑定到最终的 DF 中。

Answer 1

您可以在加入之前取消旋转（堆栈）df2：

result = df1.join(
    df2.selectExpr(
        'name',
        'stack(3, ' + ', '.join(["'%s', %s" % (c, c) for c in df2.columns[1:]]) + ') as (which_col, value)'
    ), 
    ['name', 'which_col'], 
    'left'
)

result.show()
+----+---------+---+-----+
|name|which_col| id|value|
+----+---------+---+-----+
|John|     col1|  1|   75|
|John|     col1|  1|   45|
|Jane|     col3|  2|   21|
| Bob|     col2|  3|   98|
| Bob|     col2|  3|   65|
| Bob|     col2|  3|   43|
|Barb|     col1|  4| null|
+----+---------+---+-----+