Left反加入Spark？

Question

我定义了两个表是这样的：

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) 

我试图建立查询到从没有匹配的行表2表1返回行。 我尝试使用此查询做到这一点：

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 

但这只是给了我从表1中的所有行：

列表（{“名”：“约翰”，“年龄”： 33，isBoy：true}， {“name”：“susan”，“age”：26，“isBoy”：false}， {“name”：“mike”，“age”：26，“isBoy “：true}）

如何在Spark中有效地进行这种连接？

我正在查找SQL查询，因为我需要能够指定要在两个表之间进行比较的列，而不仅仅是比较像其他推荐问题那样逐行比较。像使用减法，除了等。

Answer 1

您可以使用“左反”连接类型 - 无论是与数据帧API或SQL（数据帧API支持SQL支持一切，包括任何连接，你需要条件）：

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) 

注：它也值得注意的是，有而不单独指定模式产生的采样数据，使用元组和隐式toDF方法的更短，更简洁的方式，然后“固定”，其中所需的自动-推断出的模式：

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

Answer 2

你可以使用内置函数except （我会用你提供的代码，但你没有包括进口，所以我不能只是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| 
+----+----+----+