How to do a Typesafe Dataset Left Join in Spark Scala

Question

Background

There has been much handwringing about how Spark's Dataset API doesn't actually provide fully type safe joins, but I'm stubborn so I really wanted to try get as close as I possibly could (without using some external library).

I was able to do that with inner joins. But, I sometimes want to do a left (or right) outer join, but can't figure out how to get the type signature correct.

Question

Specifically,

How is it possible to use the Dataset.joinWith(rightDS, condition, "left") if this function doesn't return Options on either side regardless of the (left) outer join type?

It seems either it would need to return an Option on the "outer" side (right side if doing a left join) or not return unjoinable rows at all (i.e. would become an inner join instead of a true outer join). Somehow, I need to let the "outer" fields be null when I map them into my consolidated table, but if I pattern match the "outer" row as an Option, it blocks me.

user_5555637 · Answer

TL;DR

Spark's Dataset API returns a null for the entire unjoinable "outer" record.

... not a null at the field level (like SQL) or an Option at the record or field level that you could simply match. :( .... In short, the joinWith function lies about the type signature and casually returns a silent null instead of the prescribed case class (or type).

Solution

In type signatures and pattern matches, act like it's going to return a valid record (e.g. don't wrap a pattern match in Option because that won't match their type signature).
But, before actually using the field, create a new val that wraps the potentially null "outer" record in an Option.

Example

This should be valid scala code (assuming you've set up the spark context and defined the Datasets and case classes), but I haven't tested it.

val joinedDs = leftDs
  .joinWith(rightDs, leftDs("key_field") === rightDs("key_field"), "left")
  .map { x => {
    val l = x._1 // alias the left side
    val r = Option(x._2) // alias and Optionalize the right side
    ResultCaseClass( // defined elsewhere
      l.key_field,
      l.non_key_field, // may be an Option, or not - just as it appears in leftDs
      option_field = r.map(_.regular_field), // turns nonOption rightDS to Option
      reoption_field = r.flatMap(_.already_an_option) // flatten rightDs Option
    )
  }}