Gauche Anti rejoindre Étincelle?

J'ai défini deux tables comme ceci:

 val tableName = "table1"
val tableName2 = "table2"
val format = new SimpleDateFormat("yyyy-MM-dd")
val data = List(
List("mike", 26, true),
List("susan", 26, false),
List("john", 33, true)
)
val data2 = List(
List("mike", "grade1", 45, "baseball", new java.sql.Date(format.parse("1957-12-10").getTime)),
List("john", "grade2", 33, "soccer", new java.sql.Date(format.parse("1978-06-07").getTime)),
List("john", "grade2", 32, "golf", new java.sql.Date(format.parse("1978-06-07").getTime)),
List("mike", "grade2", 26, "basketball", new java.sql.Date(format.parse("1978-06-07").getTime)),
List("lena", "grade2", 23, "baseball", new java.sql.Date(format.parse("1978-06-07").getTime))
)
val rdd = sparkContext.parallelize(data).map(Row.fromSeq(_))
val rdd2 = sparkContext.parallelize(data2).map(Row.fromSeq(_))
val schema = StructType(Array(
StructField("name", StringType, true),
StructField("age", IntegerType, true),
StructField("isBoy", BooleanType, false)
))
val schema2 = StructType(Array(
StructField("name", StringType, true),
StructField("grade", StringType, true),
StructField("howold", IntegerType, true),
StructField("hobby", StringType, true),
StructField("birthday", DateType, false)
))
val df = sqlContext.createDataFrame(rdd, schema)
val df2 = sqlContext.createDataFrame(rdd2, schema2)
df.createOrReplaceTempView(tableName)
df2.createOrReplaceTempView(tableName2)

Je suis en train de construire une requête pour renvoyer les lignes de la table table1 qui n'ont pas de ligne correspondante dans la table2.
J'ai essayé de le faire en utilisant cette requête:

Select * from table1 LEFT JOIN table2 ON table1.name = table2.name AND table1.age = table2.howold AND table2.name IS NULL AND table2.howold IS NULL

mais cela me donne toutes les lignes de la table table1:

Liste({"nom":"jean","age":33,"isBoy":true},
{"name":"susan","age":26,"isBoy":false},
{"name":"mike","age":26,"isBoy":true})

Comment faire ce type de jointure dans Spark efficacement?

Je suis à la recherche pour une requête SQL car j'ai besoin d'être en mesure de spécifier les colonnes à comparer entre les deux tables, et pas seulement de comparer, ligne par ligne, comme cela se fait dans d'autres questions recommandées. Comme l'utilisation de soustraire, à l'exception etc.

Double Possible de Spark: soustraire deux DataFrames
en fonction de votre modifier et commenter ma réponse, je pense que vous êtes à la recherche de: stackoverflow.com/questions/29537564/... notamment le commentaire de @Interfector sur la première réponse
RDD#cogroup devrait fonctionner

OriginalL'auteur sergeda | 2017-04-03

  1. 18

    Vous pouvez utiliser de la "gauche anti" type de jointure - soit avec un DataFrame de l'API ou avec SQL (DataFrame API prend en charge tout ce qui SQL prend en charge, y compris toute condition de jointure dont vous avez besoin):

    DataFrame API:

    df.as("table1").join(
df2.as("table2"),
$"table1.name" === $"table2.name" && $"table1.age" === $"table2.howold",
"leftanti"
)

    SQL:

    sqlContext.sql(
"""SELECT table1.* FROM table1
| LEFT ANTI JOIN table2
| ON table1.name = table2.name AND table1.age = table2.howold
""".stripMargin)

    NOTE: il est également intéressant de noter qu'il existe un plus court, plus concis de la création de l'échantillon de données sans spécifier le schéma séparément, à l'aide de n-uplets et l'implicite toDF méthode, puis de "fixer" le automatiquement déduit du schéma en cas de besoin:

    val df = List(
("mike", 26, true),
("susan", 26, false),
("john", 33, true)
).toDF("name", "age", "isBoy")
val df2 = List(
("mike", "grade1", 45, "baseball", new java.sql.Date(format.parse("1957-12-10").getTime)),
("john", "grade2", 33, "soccer", new java.sql.Date(format.parse("1978-06-07").getTime)),
("john", "grade2", 32, "golf", new java.sql.Date(format.parse("1978-06-07").getTime)),
("mike", "grade2", 26, "basketball", new java.sql.Date(format.parse("1978-06-07").getTime)),
("lena", "grade2", 23, "baseball", new java.sql.Date(format.parse("1978-06-07").getTime))
).toDF("name", "grade", "howold", "hobby", "birthday").withColumn("birthday", $"birthday".cast(DateType))

    OriginalL'auteur Tzach Zohar

  2. 3

    Vous pouvez le faire avec la fonction intégrée except
    (Je l'aurais utilisé le code fourni par vous, mais vous n'avez pas inclure les importations, donc je ne pouvais pas juste c/p 🙁 )

    val a = sc.parallelize(Seq((1,"a",123),(2,"b",456))).toDF("col1","col2","col3")
val b= sc.parallelize(Seq((4,"a",432),(2,"t",431),(2,"b",456))).toDF("col1","col2","col3")
scala> a.show()
+----+----+----+
|col1|col2|col3|
+----+----+----+
|   1|   a| 123|
|   2|   b| 456|
+----+----+----+
scala> b.show()
+----+----+----+
|col1|col2|col3|
+----+----+----+
|   4|   a| 432|
|   2|   t| 431|
|   2|   b| 456|
+----+----+----+
scala> a.except(b).show()
+----+----+----+
|col1|col2|col3|
+----+----+----+
|   1|   a| 123|
+----+----+----+
    Je suis à la recherche pour une requête SQL car j'ai besoin d'être en mesure de spécifier les colonnes à comparer entre les deux tables, et pas seulement de comparer, ligne par ligne

    OriginalL'auteur James Tobin