CODEWITHSUNDEEP
dataframescalaapache-spark
Nab
Nab

Reputation: 137

Scala MatchError while joining a dataframe and a dataset

I have one dataframe and one dataset :

Dataframe 1 :

+------------------------------+-----------+
|City_Name                     |Level      |
+------------------------------+------------
|{City -> Paris}               |86         |       
+------------------------------+-----------+

Dataset 2 :

+-----------------------------------+-----------+
|Country_Details                    |Temperature|
+-----------------------------------+------------
|{City -> Paris, Country -> France} |31         |       
+-----------------------------------+-----------+

I am trying to make a join of them by checking if the map in the column "City_Name" is included in the map of the Column "Country_Details".

I am using the following UDF to check the condition :

    val mapEqual = udf((col1: Map[String, String], col2: Map[String, String]) => {
        if (col2.nonEmpty){
            col2.toSet subsetOf col1.toSet
        } else {
            true
        }
    })

And I am making the join this way :

dataset2.join(dataframe1 , mapEqual(dataset2("Country_Details"), dataframe1("City_Name"), "leftanti")

However, I get such error :

terminated with error scala.MatchError: UDF(Country_Details#528) AS City_Name#552 (of class org.apache.spark.sql.catalyst.expressions.Alias)

Has anyone previously got the same error ? I am using Spark version 3.0.2 and SQLContext, with scala language.

Upvotes: 0

Views: 172

Answers (1)

AminMal
AminMal

Reputation: 3183

There are 2 issues here, the first one is that when you're calling your function, you're passing one extra parameter leftanti (you meant to pass it to join function, but you passed it to the udf instead). The second one is that the udf logic won't work as expected, I suggest you use this:

val mapContains = udf { (col1: Map[String, String], col2: Map[String, String]) => 
  col2.keys.forall { key => 
    col1.get(key).exists(_ eq col2(key))
  }
}

Result:

scala> ds.join(df1 , mapContains(ds("Country_Details"), df1("City_Name")), "leftanti").show(false)
+----------------------------------+-----------+
|Country_Details                   |Temperature|
+----------------------------------+-----------+
|{City -> Paris, Country -> France}|31         |
+----------------------------------+-----------+

Upvotes: 1

Related Questions