Spark especificar múltiples condiciones de columna para la unión de dataframe

A partir de la versión 1.5.0 de Spark (que actualmente no se ha publicado), puede unirse en varias columnas de DataFrame. Refiérase a SPARK-7990: Agregue métodos para facilitar equi-join en múltiples claves de unión .

Python

Leads.join(
    Utm_Master, 
    ["LeadSource","Utm_Source","Utm_Medium","Utm_Campaign"],
    "left_outer"
)

Scala

La pregunta pedía una respuesta Scala, pero no uso Scala. Esta es mi mejor suposición....

Leads.join(
    Utm_Master,
    Seq("LeadSource","Utm_Source","Utm_Medium","Utm_Campaign"),
    "left_outer"
)

Una cosa que puede hacer es usar SQL raw:

case class Bar(x1: Int, y1: Int, z1: Int, v1: String)
case class Foo(x2: Int, y2: Int, z2: Int, v2: String)

val bar = sqlContext.createDataFrame(sc.parallelize(
    Bar(1, 1, 2, "bar") :: Bar(2, 3, 2, "bar") ::
    Bar(3, 1, 2, "bar") :: Nil))

val foo = sqlContext.createDataFrame(sc.parallelize(
    Foo(1, 1, 2, "foo") :: Foo(2, 1, 2, "foo") ::
    Foo(3, 1, 2, "foo") :: Foo(4, 4, 4, "foo") :: Nil))

foo.registerTempTable("foo")
bar.registerTempTable("bar")

sqlContext.sql(
    "SELECT * FROM foo LEFT JOIN bar ON x1 = x2 AND y1 = y2 AND z1 = z2")

En Pyspark simplemente puede especificar cada condición por separado:

val Lead_all = Leads.join(Utm_Master,  
    (Leaddetails.LeadSource == Utm_Master.LeadSource) &
    (Leaddetails.Utm_Source == Utm_Master.Utm_Source) &
    (Leaddetails.Utm_Medium == Utm_Master.Utm_Medium) &
    (Leaddetails.Utm_Campaign == Utm_Master.Utm_Campaign))

Solo asegúrese de usar operadores y paréntesis correctamente.

Scala:

Leaddetails.join(
    Utm_Master, 
    Leaddetails("LeadSource") <=> Utm_Master("LeadSource")
        && Leaddetails("Utm_Source") <=> Utm_Master("Utm_Source")
        && Leaddetails("Utm_Medium") <=> Utm_Master("Utm_Medium")
        && Leaddetails("Utm_Campaign") <=> Utm_Master("Utm_Campaign"),
    "left"
)

Para hacerlo insensible a mayúsculas y minúsculas ,

import org.apache.spark.sql.functions.{lower, upper}

Entonces simplemente use lower(value) en la condición del método join.

Eg: dataFrame.filter(lower(dataFrame.col("vendor")).equalTo("fortinet"))

Las opciones === me dan columnas duplicadas. Así que uso Seq en su lugar.

val Lead_all = Leads.join(Utm_Master,
    Seq("Utm_Source","Utm_Medium","Utm_Campaign"),"left")

Por supuesto, esto solo funciona cuando los nombres de las columnas de unión son los mismos.

Spark SQL soporta join en tupla de columnas cuando está entre paréntesis, como

... WHERE (list_of_columns1) = (list_of_columns2)

Que es una forma más corta que especificar expresiones iguales ( = ) para cada par de columnas combinadas por un conjunto de "Y"s.

Por ejemplo:

SELECT a,b,c
FROM    tab1 t1
WHERE 
   NOT EXISTS
   (    SELECT 1
        FROM    t1_except_t2_df e
        WHERE (t1.a, t1.b, t1.c) = (e.a, e.b, e.c)
   )

En lugar de

SELECT a,b,c
FROM    tab1 t1
WHERE 
   NOT EXISTS
   (    SELECT 1
        FROM    t1_except_t2_df e
        WHERE t1.a=e.a AND t1.b=e.b AND t1.c=e.c
   )

Que también es menos legible, especialmente cuando la lista de columnas es grande y desea tratar con NULLs fácilmente.