Spark gibt mehrere Spaltenbedingungen für den Dataframe-Join an

Question

Wie gibt man mehr Spaltenbedingungen vor, wenn zwei Datenrahmen verbunden werden. Zum Beispiel möchte ich Folgendes ausführen:

val Lead_all = Leads.join(Utm_Master, 
    Leaddetails.columns("LeadSource","Utm_Source","Utm_Medium","Utm_Campaign") == 
    Utm_Master.columns("LeadSource","Utm_Source","Utm_Medium","Utm_Campaign"), 
"left") 

Ich möchte nur beitreten, wenn diese Spalten übereinstimmen. Aber obige Syntax ist nicht gültig, da Spalten nur eine Zeichenkette benötigt. Wie bekomme ich, was ich will?

Answer 1

Es wird ein Funke column/expression API join für einen solchen Fall:

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" 
) 

Der <=> Operator im Beispiel bedeutet "Equality test that is safe for null values".

Der Hauptunterschied zu einfachen Equality test (===) ist, dass die erste sicher verwendet werden kann, wenn eine der Spalten Nullwerte haben darf.

Answer 2

Eine Sache, die Sie tun können, ist roh SQL zu verwenden:

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") 

Answer 3

Ab der Spark-Version 1.5.0 (die derzeit nicht veröffentlicht ist) können Sie an mehreren DataFrame-Spalten teilnehmen. Siehe SPARK-7990: Add methods to facilitate equi-join on multiple join keys.

Python

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

Scala

Die Frage nach einer Antwort Scala gefragt, aber ich verwende Scala nicht. Hier ist meine beste Vermutung ....

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

Answer 4

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" 
) 

Um es Groß- und Kleinschreibung, machen

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

dann benutzen Sie einfach lower(value) im Zustand der Join-Methode.

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

Answer 5

In Pyspark Sie einfach jede Bedingung separat angeben:

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)) 

Nur sicher sein, richtig Operatoren und Klammern zu verwenden.

Answer 6

Funken SQL unterstützt auf Tupel von Spalten, wenn in Klammern verbinden, wie

... WHERE (list_of_columns1) = (list_of_columns2) 

, die einen Weg kürzer als Angabe gleich Ausdrücke (=) für jedes Paar von Spalten, die durch einen Satz von „und“ s kombinierte .

Zum Beispiel:

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) 
    ) 

statt

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 
    ) 

, die zu weniger lesbar ist vor allem, wenn die Liste der Spalten groß ist und Sie wollen einfach mit NULL-Werten beschäftigen.