Как сравнить столбцы данных Spark с другими значениями столбцов данных

Question

У меня есть два кадра данных, как показано ниже:

df1: Which will have few values as below. This is dynamic.

+--------------+
|tags          |
+--------------+
|first_name    |
|last_name     |
|primary_email |
|other_email   |
+--------------+

df2: второй кадр данных имеет несколько предопределенных комбинаций, как показано ниже:

+---------------------------------------------------------------------------------------------+
|combinations                                                                                 |
+---------------------------------------------------------------------------------------------+
|last_name, first_name, primary_email                                                         |
|last_name, first_name, other_email                                                           |
|last_name, primary_email, primary_phone                                                      |
|last_name, primary_email, secondary_phone                                                    |
|last_name, address_line1, address_line2,city_name, state_name,postal_code, country_code, guid|
+---------------------------------------------------------------------------------------------+

Ожидаемый результат DF: Теперь я хотел чтобы найти это из моего кадра данных, есть ли какие-либо действительные комбинации, которые я могу сделать. Результат должен иметь все допустимые комбинации, если он совпадает с любой с combinations dataframe.

resultDF:

+---------------------------------------------------------------------------------------------+
|combinations                                                                                 |
+---------------------------------------------------------------------------------------------+
|last_name, first_name, primary_email                                                         |
|last_name, first_name, other_email                                                           |
+---------------------------------------------------------------------------------------------+

Я попытался преобразовать оба кадра данных в список и попытаться сравнить их, но я всегда получаю 0 комбинаций .

Код scala, который я пробовал.

val combinationList = combinations.map(r => r.getString(0)).collect.toList

var combList: Seq[Seq[String]]  = Seq.empty

    for (comb <- combinationList) {
      var tmp: Seq[String] = Seq.empty
      tmp = tmp :+ comb
      combList = combList :+ tmp
    }

val result = combList.filter(
      list => df1.filter(df1.col("tags").isin(list: _*)).count == list.size
    )

println(result.size)

Это всегда возвращает 0. Ответ должен быть 2.

Может кто-нибудь подсказать мне, что лучше подход?

Answer 1

Попробуйте это. Соберите свой df1, создайте новый столбец массива в df2 со значениями df1. Сравните два массива, используя array_except, если используете Spark 2.4, который возвращает разницу массивов буксировки. Затем отфильтруйте, если размер этого == 0

scala> val df1 = Seq(
     |   "first_name",
     |   "last_name",
     |   "primary_email",
     |   "other_email" 
     | ).toDF("tags")
df1: org.apache.spark.sql.DataFrame = [tags: string]

scala> 

scala> val df2 = Seq(
     | Seq("last_name", "first_name", "primary_email"),                                                         
     | Seq("last_name", "first_name", "other_email"),
     | Seq("last_name", "primary_email", "primary_phone"),                                                      
     | Seq("last_name", "primary_email", "secondary_phone"),
     | Seq("last_name", "address_line1", "address_line2", "city_name", "state_name", "postal_code", "country_code", "guid")
     | ).toDF("combinations")
df2: org.apache.spark.sql.DataFrame = [combinations: array<string>]

scala> 

scala> df1.show(false)
+-------------+
|tags         |
+-------------+
|first_name   |
|last_name    |
|primary_email|
|other_email  |
+-------------+


scala> 

scala> df2.show(false)
+-------------------------------------------------------------------------------------------------+
|combinations                                                                                     |
+-------------------------------------------------------------------------------------------------+
|[last_name, first_name, primary_email]                                                           |
|[last_name, first_name, other_email]                                                             |
|[last_name, primary_email, primary_phone]                                                        |
|[last_name, primary_email, secondary_phone]                                                      |
|[last_name, address_line1, address_line2, city_name, state_name, postal_code, country_code, guid]|
+-------------------------------------------------------------------------------------------------+


scala> 

scala> val df1tags = df1.collect.map(r => r.getString(0))
df1tags: Array[String] = Array(first_name, last_name, primary_email, other_email)

scala> 

scala> val df3 = df2.withColumn("tags", lit(df1tags))
df3: org.apache.spark.sql.DataFrame = [combinations: array<string>, tags: array<string>]

scala> df3.show(false)
+-------------------------------------------------------------------------------------------------+---------------------------------------------------+
|combinations                                                                                     |tags                                               |
+-------------------------------------------------------------------------------------------------+---------------------------------------------------+
|[last_name, first_name, primary_email]                                                           |[first_name, last_name, primary_email, other_email]|
|[last_name, first_name, other_email]                                                             |[first_name, last_name, primary_email, other_email]|
|[last_name, primary_email, primary_phone]                                                        |[first_name, last_name, primary_email, other_email]|
|[last_name, primary_email, secondary_phone]                                                      |[first_name, last_name, primary_email, other_email]|
|[last_name, address_line1, address_line2, city_name, state_name, postal_code, country_code, guid]|[first_name, last_name, primary_email, other_email]|
+-------------------------------------------------------------------------------------------------+---------------------------------------------------+


scala> 

scala> val df4 = df3.withColumn("combMinusTags", array_except($"combinations", $"tags"))
df4: org.apache.spark.sql.DataFrame = [combinations: array<string>, tags: array<string> ... 1 more field]

scala> df4.show(false)
+-------------------------------------------------------------------------------------------------+---------------------------------------------------+--------------------------------------------------------------------------------------+
|combinations                                                                                     |tags                                               |combMinusTags                                                                         |
+-------------------------------------------------------------------------------------------------+---------------------------------------------------+--------------------------------------------------------------------------------------+
|[last_name, first_name, primary_email]                                                           |[first_name, last_name, primary_email, other_email]|[]                                                                                    |
|[last_name, first_name, other_email]                                                             |[first_name, last_name, primary_email, other_email]|[]                                                                                    |
|[last_name, primary_email, primary_phone]                                                        |[first_name, last_name, primary_email, other_email]|[primary_phone]                                                                       |
|[last_name, primary_email, secondary_phone]                                                      |[first_name, last_name, primary_email, other_email]|[secondary_phone]                                                                     |
|[last_name, address_line1, address_line2, city_name, state_name, postal_code, country_code, guid]|[first_name, last_name, primary_email, other_email]|[address_line1, address_line2, city_name, state_name, postal_code, country_code, guid]|
+-------------------------------------------------------------------------------------------------+---------------------------------------------------+--------------------------------------------------------------------------------------+


scala> 

scala> 

scala> df4.filter(size($"combMinusTags") === 0).show(false)
+--------------------------------------+---------------------------------------------------+-------------+
|combinations                          |tags                                               |combMinusTags|
+--------------------------------------+---------------------------------------------------+-------------+
|[last_name, first_name, primary_email]|[first_name, last_name, primary_email, other_email]|[]           |
|[last_name, first_name, other_email]  |[first_name, last_name, primary_email, other_email]|[]           |
+--------------------------------------+---------------------------------------------------+-------------+

Spark 2.3

, напишите свою собственную функцию array_except как udf.

scala> def array_expt[T](a: Seq[T], b:Seq[T]):Seq[T] = {
     |   a.diff(b)
     | } 
array_expt: [T](a: Seq[T], b: Seq[T])Seq[T]

scala> 

scala> val myUdf = udf { array_expt[String] _ }
myUdf: org.apache.spark.sql.expressions.UserDefinedFunction = UserDefinedFunction(<function2>,ArrayType(StringType,true),Some(List(ArrayType(StringType,true), ArrayType(StringType,true))))

scala> 

scala> val df4 = df3.withColumn("combMinusTags", myUdf($"combinations", $"tags"))
df4: org.apache.spark.sql.DataFrame = [combinations: array<string>, tags: array<string> ... 1 more field]

scala> df4.show(false)
+-------------------------------------------------------------------------------------------------+---------------------------------------------------+--------------------------------------------------------------------------------------+
|combinations                                                                                     |tags                                               |combMinusTags                                                                         |
+-------------------------------------------------------------------------------------------------+---------------------------------------------------+--------------------------------------------------------------------------------------+
|[last_name, first_name, primary_email]                                                           |[first_name, last_name, primary_email, other_email]|[]                                                                                    |
|[last_name, first_name, other_email]                                                             |[first_name, last_name, primary_email, other_email]|[]                                                                                    |
|[last_name, primary_email, primary_phone]                                                        |[first_name, last_name, primary_email, other_email]|[primary_phone]                                                                       |
|[last_name, primary_email, secondary_phone]                                                      |[first_name, last_name, primary_email, other_email]|[secondary_phone]                                                                     |
|[last_name, address_line1, address_line2, city_name, state_name, postal_code, country_code, guid]|[first_name, last_name, primary_email, other_email]|[address_line1, address_line2, city_name, state_name, postal_code, country_code, guid]|
+-------------------------------------------------------------------------------------------------+---------------------------------------------------+--------------------------------------------------------------------------------------+


scala> 

scala> df4.filter(size($"combMinusTags") === 0).show(false)
+--------------------------------------+---------------------------------------------------+-------------+
|combinations                          |tags                                               |combMinusTags|
+--------------------------------------+---------------------------------------------------+-------------+
|[last_name, first_name, primary_email]|[first_name, last_name, primary_email, other_email]|[]           |
|[last_name, first_name, other_email]  |[first_name, last_name, primary_email, other_email]|[]           |
+--------------------------------------+---------------------------------------------------+-------------+