apache-spark - 在 pyspark 中读取镶木地板文件时 where 子句中的动态列

Question

我有 parquet 文件，想根据动态列读取它们，所以举个例子，我有 2 个数据框，想根据 df2 从 df1 中选择数据。

所以我正在使用下面的代码，但希望在加入列方面使其动态化，今天我有 2 个列，明天我可以有 4 个。

a = dict[keys]
col1 = a[0]
col2 = a[1]
v = df1.join(df2,[df1[col1] == df2[col1],
df1.[col2] == df2.[col2],
how='inner')

那么如何使这些列动态化，以便不需要对连接条件进行硬编码，并将添加从连接条件中删除的列。

标签： apache-sparkdataframedynamicpyspark

Answer 1

我会首先根据您的 dict 生成一个连接条件对象，然后在连接中使用它。

from functools import reduce

join_condition = reduce(
    lambda a, b: a && b,
    [ df1[col] == df2[col] 
     for col 
     in dict[keys]
    ]
)

v = df1.join(
    df2,
    join_condition,
)