如何在 Spark Java API 中编写动态连接条件

Question

我想使用 spark Java API 对数据集执行左外连接。如何编写动态条件以匹配连接条件中的多个列。

我有两个数据集对象。它们都有 2 列或更多列。我无法定义条件

将一列与另一列匹配的示例

dataSet = resultData.as("resultData").join(distinctData.as("distinctData"), resultData.col("A").equalTo(distinctData.col("B")), "leftouter").selectExpr(select.toString());

现在由于有多个列，我无法使用 Java API 定义动态表达式来匹配多个列。

Answer 1

未经测试的代码 - 但这会从列名列表动态生成连接条件

public Column makeJoinConditional(Dataset<Row> df1, Dataset<Row> df2, List<String> columnNames, Column c)  {

        if (c==null) {
            String  top = columnNames.get(0);
            columnNames.remove(0);
            Column first = df1.col(top).equalTo(df2.col(top));

            return makeJoinConditional(df1,df2, columnNames,first);

        } else {

            if (columnNames.size()==0) {
                return c;
            } else {
                String  top = columnNames.get(0);
                columnNames.remove(0);
                Column next = c.and( df1.col(top).equalTo(df2.col(top)) );
                return makeJoinConditional(df1,df2, columnNames,next);
            }
        }
    }

    public Dataset<Row> joinDataFrames(Dataset<Row> df1, Dataset<Row> df2, List<String> columns) {
        Column joinCols = makeJoinConditional(df1,df2,columns,null);
        return df1.join(df2,joinCols);
    }