Объединение двух фреймов данных в Spark Scala на основе условия ИЛИ

Question

У меня есть два фрейма данных: 1) Учетные записи и 2) Клиенты. Схема счетов такая:

Name  Id    Telehone     Mob       email
 AR     1     123        1234      [e-mail]
 BR     2     213        4123      [e-mail]
 CR     3     231        3214      [e-mail]
 KR     4     132        1324      [e-mail]

Второй стол Клиенты как:

Id    Phone   Email
  2     2344    [e-mail]
  6     132     [e-mail]
  7     64562    [e-mail]

Мне нужно объединить эти два фрейма данных так, чтобы Id соответствует Id OR Phone соответствует Telephone OR Mob Or Email соответствует email. В приведенном выше случае в первой строке клиентов соответствует идентификатор, второй соответствует по телефону, а третий - по электронной почте. Объединение следует оставить во всех записях учетных записей.

Answer 1

Проверьте код ниже.

scala> accountDF.show(false)
+----+---+---------+----+---------------+
|name|id |telephone|mob |email          |
+----+---+---------+----+---------------+
|AR  |1  |123      |1234|[e-mail]|
|BR  |2  |213      |4123|[e-mail]|
|CR  |3  |231      |3214|[e-mail]|
|KR  |4  |132      |1324|[e-mail]|
+----+---+---------+----+---------------+

scala> customerDF.show(false)
+---+-----+---------------+
|id |phone|email          |
+---+-----+---------------+
|2  |2344 |[e-mail]|
|6  |132  |[e-mail]|
|7  |64562|[e-mail]|
+---+-----+---------------+

scala> accountDF.printSchema
root
 |-- name: string (nullable = true)
 |-- id: string (nullable = true)
 |-- telephone: string (nullable = true)
 |-- mob: string (nullable = true)
 |-- email: string (nullable = true)

scala> customerDF.printSchema
root
 |-- id: string (nullable = true)
 |-- phone: string (nullable = true)
 |-- email: string (nullable = true)

scala> 

accountDF.join(customerDF,(accountDF("id") === customerDF("id") || (accountDF("telephone") === customerDF("phone") ||accountDF("mob") === customerDF("phone")) || accountDF("email") === customerDF("email")),"left").show(false)

+----+---+---------+----+---------------+----+-----+---------------+
|name|id |telephone|mob |email          |id  |phone|email          |
+----+---+---------+----+---------------+----+-----+---------------+
|AR  |1  |123      |1234|[e-mail]|7   |64562|[e-mail]|
|BR  |2  |213      |4123|[e-mail]|2   |2344 |[e-mail]|
|CR  |3  |231      |3214|[e-mail]|null|null |null           |
|KR  |4  |132      |1324|[e-mail]|6   |132  |[e-mail]|
+----+---+---------+----+---------------+----+-----+---------------+

Answer 2

Вы можете легко удовлетворить это требование с помощью spark SQL.

Код для ссылки -

import org.apache.spark.sql.functions._

val accountdf = sc.parallelize(Seq(("AR",1,123,1234,"[e-mail]"),("BR", 2, 213, 4123, "[e-mail]"),("CR", 3, 231, 3214, "[e-mail]"),("KR", 4, 132, 1324, "[e-mail]"))).toDF("name","id","telephone","mob","email")


accountdf.createOrReplaceTempView("account")

val customerdf = sc.parallelize(Seq((2,2344,"[e-mail]"),(6,132,"[e-mail]"),(7,64562,"[e-mail]"))).toDF("id","phone","email")
   
customerdf.createOrReplaceTempView("customer")

sql("select * from account a left join customer c on a.id = c.id or (a.telephone = c.phone or a.mob = c.phone) or a.email = c.email").show(false)

+----+---+---------+----+---------------+----+-----+---------------+
|name|id |telephone|mob |email          |id  |phone|email          |
+----+---+---------+----+---------------+----+-----+---------------+
|BR  |2  |213      |4123|[e-mail]|2   |2344 |[e-mail]|
|KR  |4  |132      |1324|[e-mail]|6   |132  |[e-mail]|
|AR  |1  |123      |1234|[e-mail]|7   |64562|[e-mail]|
|CR  |3  |231      |3214|[e-mail]|null|null |null           |
+----+---+---------+----+---------------+----+-----+---------------+

Answer 3

  val sourceDF = Seq(("AR",1,123,1234,"[e-mail]"),
    ("BR",2,213,4123,"[e-mail]"),
  ("CR",3,231,3214,"[e-mail]"),
  ("KR",4,132,1324,"[e-mail]")
  ).toDF("Name","Id","Telehone","Mob","email")


  val sourceDF2 = Seq((2,2344,"[e-mail]"),
    (6,132,"[e-mail]"),
    (7,64562,"[e-mail]")
  ).toDF("Id","Phone","Email")

  val joinDF = sourceDF.join(sourceDF2,
    sourceDF.col("Id") === sourceDF2.col("Id") ||
      (sourceDF.col("Telehone") === sourceDF2.col("Phone") ||
      sourceDF.col("Mob") === sourceDF2.col("Phone")) ||
      sourceDF.col("email") === sourceDF2.col("Email")
    ,
  "inner")
  // use "inner" or "left" or ...