How to write dynamic join condition in spark Java API

Question

I want to perform left outer join on Dataset using spark Java API. How to write dynamic condition to match the multiple columns in join condition.

I am having two dataset objects. Both of them having 2 or more columns. I am not able to define condition

Example which match 1 column with another

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

Now Since there are multiple column I am not able to define dynamic expression for matching the multiple columns using Java API.

you probably got a downvote because you haven't included anything about what your data looks like, or what you've tried so far. I'd be happy to help you if you could provide that information. — nont
– nont, Commented Apr 23, 2019 at 13:25
do you get an error? what happens when you run the code above? — nont
– nont, Commented Apr 23, 2019 at 14:05
for example mentioned in the question i don't get any error. Issue is I want to specify condition for matching multiple columns and I am not able to find any reference to define the same. — VVP
– VVP, Commented Apr 23, 2019 at 14:12

nont · Accepted Answer · 2019-04-23 15:56:18Z

3

Untested code - but this dynamically generates a join condition from a list of column names

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);
    }

edited Apr 23, 2019 at 15:56

answered Apr 23, 2019 at 14:49

nont

9,5597 gold badges65 silver badges83 bronze badges

Sign up to request clarification or add additional context in comments.

4 Comments

VVP Over a year ago

Yes, This will work when am having fixed number of columns. But column number varies with each scenario and I can't change code every time for this :)

nont Over a year ago

so, you need a function, that given a list of columns, can generate the conditional statement?

VVP Over a year ago

Yes, Somewhat similar to you said. But dataste.join() accepts column, columnExpr and seq. I am trying to find what is columnExpr which will return the conditional column/ statement.

nont Over a year ago

ok, updated answer to dynamically generate conditional based on a list of column names

Collectives™ on Stack Overflow

How to write dynamic join condition in spark Java API

1 Answer 1

4 Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

4 Comments

Your Answer

Sign up or log in

Post as a guest

Related