Spark SQL 的 where 子句排除空值

Question

我正在尝试在Apache spark sql上运行查询。第一个查询运行良好，但第二个查询也会删除空值。

代码：

def main(args: Array[String]) {
    val sc = new SparkContext("local[*]", "Spark")
    val sqlContext = new SQLContext(sc)
    val pageViewsDF = getDataframe(sc, sqlContext)
    println("RUNNING SQL QUERIES ")
    sqlContext.sql("select name , count(*) from pageviews_by_second group by name").show(10)
    sqlContext.sql("select name , count(*) from pageviews_by_second where name not in ("Rose") group by name").show(10)
  }
  def getDataframe(sc: SparkContext, sqlContext: SQLContext): DataFrame = {
    Logger.getLogger("org").setLevel(Level.OFF);
    Logger.getLogger("akka").setLevel(Level.OFF);
    val dataArray = List(List("David", null),
      List("David", null),
      List("Charlie", "23"),
      List("Rose", null),
      List("Ben", null),
      List("Harry", "43"),
      List(null, "25"),
      List(null, "21"),
      List("David", "15"),
      List("Rose", null),
      List("Alan", "26"))
    val separator = ","
    // Create an RDD
    val dataRDD = sc.parallelize(dataArray)
    // The schema is encoded in a string
    val header = "name,age"
    // Import Spark SQL data types and Row.
    import org.apache.spark.sql._
    // Generate the schema based on the string of schema
    val schema =
      StructType(
        header.split(separator).map { fieldName =>
          StructField(fieldName, StringType, true)
        })
    val rowRDD =
      dataRDD
        .map(p => Row(p(0), p(1)))
    // Apply the schema to the RDD.
    var df = sqlContext.createDataFrame(rowRDD, schema)
    df.registerTempTable("pageviews_by_second")
    df
  }

第一次查询的结果是：

+-------+---+
|   name|_c1|
+-------+---+
|   Alan|  1|
|    Ben|  1|
|  David|  3|
|Charlie|  1|
|   Rose|  2|
|  Harry|  1|
|   null|  2|
+-------+---+

第二个查询的输出：

+-------+---+
|   name|_c1|
+-------+---+
|   Alan|  1|
|    Ben|  1|
|  David|  3|
|Charlie|  1|
|  Harry|  1|
+-------+---+

在第二个查询中，我只排除了"Rose"，但也排除了"null"。

如果我的查询是错误的，请帮助我进行正确的查询。

Answer 1

之所以会发生这种情况，是因为SQL中的NULL相当于"未知"。这意味着除了IS NULL/IS NOT NULL之外，任何与NULL的比较都是未定义的，并返回NULL。

case class Record(id: Integer, value: String)
val df = sc.parallelize(Seq(Record(1, "foo"), Record(2, null))).toDF
df.registerTempTable("df")
sqlContext.sql("""SELECT value = "foo" FROM df""").show
// +----+
// | _c0|
// +----+
// |true|
// |null|
// +----+
sqlContext.sql("""SELECT value != "foo" FROM df""").show
// +-----+
// |  _c0|
// +-----+
// |false|
// | null|
// +-----+

因为IN/NOT IN也是未定义的：

sqlContext.sql("""SELECT value IN ("foo", "bar")  FROM df""").show
// +----+
// | _c0|
// +----+
// |true|
// |null|
// +----+

这是一个标准的SQL行为，正确实现SQL标准的系统应该以相同的方式进行操作。如果你要过滤并保留NULLs，你必须明确：

sqlContext.sql(
  """SELECT value IN ("foo", "bar") OR value IS NULL FROM df""").show
// +----+
// | _c0|
// +----+
// |true|
// |true|
// +----+

Answer 2

如果您想切换回两状态逻辑，您需要将可为null的列（这是专有NVL的标准术语）coalesce转换为某个不存在的值。

使用与@zero323 相同的设置

 sqlContext.sql("""SELECT value, coalesce(value,'other baz') = "foo" FROM df""").show

value c1
foo   true
null  false