Spark Scala dataframe join dynamically using list of columns and joinExprs

Question

I am creating a function which takes join keys and condition as parameters and dynamically joins two dataframes.

I understand Spark Scala Dataframe join done the following ways:

1) join(right: Dataset[_]): DataFrame
2) join(right: Dataset[_], usingColumn: String): DataFrame
3) join(right: Dataset[_], usingColumns: Seq[String]): DataFrame
4) join(right: Dataset[_], usingColumns: Seq[String], joinType: String): DataFrame
5) join(right: Dataset[_], joinExprs: Column): DataFrame
6) join(right: Dataset[_], joinExprs: Column, joinType: String): DataFrame

Join keys/usingColumns parameters will be a list of column names. condition/joinExprs - not sure how to pass it, but it can be a string like "df2(colname) == 'xyz'"

Based on this post, I came up with the below. It takes care of join keys list, but how can I add the conditon as well? (note: I used identical dataframes here for simplicity)

 %scala
  val emp = Seq((1,"Smith",-1,"2018","10","M",3000),
    (2,"Rose",1,"2010","20","M",4000),
  )
  val empColumns = Seq("emp_id","name","superior_emp_id","year_joined","dept_id","gender","salary")
  import spark.sqlContext.implicits._
  val empDF = emp.toDF(empColumns:_*) 
  val empDF2 = emp.toDF(empColumns:_*) 


val join_keys = Seq("emp_id","name") // this will be a parameter
val joinExprs = join_keys.map{case (c1) => empDF(c1) === empDF2(c1)}.reduce(_ && _) 

// How do I add to joinExprs, another joinExpr like "empDF2(dept_id) == 10" here?

empDF.join(empDF2,joinExprs,"inner").show(false)

Raphael Roth · Accepted Answer · 2021-08-19 08:34:02Z

2

You can just append to joinExprs with &&:

empDF.join(empDF2,joinExprs && empDF2("dept_id") === 10,"inner").show(false)

answered Aug 19, 2021 at 8:34

Raphael Roth

27.3k19 gold badges98 silver badges152 bronze badges

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

3 Comments

Gadam Over a year ago

Thanks Raphael. Will try that. What if I want to send the joinExprs && "empDF2("dept_id") === 10" as a parameter, what type would that be.

Raphael Roth Over a year ago

it should be of type Column

Gadam Over a year ago

so will it be like val joinExprCol = empDF2("dept_id") === 10 looks a little strange

Collectives™ on Stack Overflow

Spark Scala dataframe join dynamically using list of columns and joinExprs

1 Answer 1

3 Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

3 Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related