字段中的空值会生成MatchError

Question

以下是有趣的：

val rddSTG = sc.parallelize(
      List ( ("RTD","ANT","SOYA BEANS", "20161123", "20161123", 4000, "docid11", null, 5) , 
             ("RTD","ANT","SOYA BEANS", "20161124", "20161123", 6000, "docid11",  null, 4) ,
             ("RTD","ANT","BANANAS", "20161124", "20161123", 7000, "docid11", null, 9) ,    
             ("HAM","ANT","CORN", "20161123", "20161123", 1000, "docid22", null, 33),
             ("LIS","PAR","BARLEY", "20161123", "20161123", 11111, "docid33", null, 44)
           )
                          )

val dataframe = rddSTG.toDF("ORIG", "DEST", "PROD", "PLDEPDATE", "PLARRDATE", "PLCOST", "docid", "ACTARRDATE", "mutationseq")
dataframe.createOrReplaceTempView("STG")
spark.sql("SELECT * FROM STG ORDER BY PLDEPDATE DESC").show()

它会生成如下错误：

scala.MatchError: Null (of class scala.reflect.internal.Types$TypeRef$$anon$6)

只要我将其中一个空值更改为非空值，它就会工作。我想我明白了，因为在这个领域无法做出任何推断，但这确实看起来很奇怪。想法？

Answer 1

简单的解决方案是添加测试行

val rddSTG = sc.parallelize(
  Seq ( ("RTD","ANT","SOYA BEANS", "20161123", "20161123", 4000, "docid11", null, 5) ,
    ("RTD","ANT","SOYA BEANS", "20161124", "20161123", 6000, "docid11",  null, 4) ,
    ("RTD","ANT","BANANAS", "20161124", "20161123", 7000, "docid11", null, 9) ,
    ("HAM","ANT","CORN", "20161123", "20161123", 1000, "docid22", null, 33),
    ("LIS","PAR","BARLEY", "20161123", "20161123", 11111, "docid33", null, 44),
    ("test","test","test", "test", "test", 0, "test", "", 0)
    )
    )

和过滤器在dataframe被创建为之后的测试行

val dataframe = rddSTG.toDF("ORIG", "DEST", "PROD", "PLDEPDATE", "PLARRDATE", "PLCOST", "docid", "ACTARRDATE", "mutationseq")
                      .filter(!(col("ORIG") === "test"))

您可以将其余逻辑应用为

dataframe.createOrReplaceTempView("STG")
spark.sql("SELECT * FROM STG ORDER BY PLDEPDATE DESC").show()

您应该将输出设置为

+----+----+----------+---------+---------+------+-------+----------+-----------+
|ORIG|DEST|      PROD|PLDEPDATE|PLARRDATE|PLCOST|  docid|ACTARRDATE|mutationseq|
+----+----+----------+---------+---------+------+-------+----------+-----------+
| RTD| ANT|SOYA BEANS| 20161124| 20161123|  6000|docid11|      null|          4|
| RTD| ANT|   BANANAS| 20161124| 20161123|  7000|docid11|      null|          9|
| RTD| ANT|SOYA BEANS| 20161123| 20161123|  4000|docid11|      null|          5|
| HAM| ANT|      CORN| 20161123| 20161123|  1000|docid22|      null|         33|
| LIS| PAR|    BARLEY| 20161123| 20161123| 11111|docid33|      null|         44|
+----+----+----------+---------+---------+------+-------+----------+-----------+

我希望这能有所帮助

Answer 2

我不太确定错误背后的原因，但我猜测它的发生是因为Null不能是dataframe列的数据类型。因为您的第二列是null，这是trait的一部分。由于它们位于层次结构的底部，因此不能实例化为任何其他类型。但是，null是所有内容的子类型，因此，即使您将其中任何一个更改为字符串，也会使列变为字符串类型。这只是一个假设。

但是，对于您的案例，定义一个案例类就可以了。

val rdd = sc.parallelize(List ( ("RTD","ANT","SOYA BEANS", "20161123", "20161123", 4000, "docid11", null, 5) , 
            ("RTD","ANT","SOYA BEANS", "20161124", "20161123", 6000, "docid11",  null, 4) ,
            ("RTD","ANT","BANANAS", "20161124", "20161123", 7000, "docid11", null, 9) ,    
            ("HAM","ANT","CORN", "20161123", "20161123", 1000, "docid22", null, 33),
            ("LIS","PAR","BARLEY", "20161123", "20161123", 11111, "docid33", null, 44)))
case class df_schema (ORIG: String, DEST: String, PROD: String, PLDEPDATE:String, PLARRDATE: String, PLCOSTDATE: Int, DOCID: String, ACTARRDATE: String, MUTATIONSEQ: Int)
val rddSTG = rdd.map( x=> df_schema(x._1, x._2, x._3, x._4, x._5, x._6, x._7, x._8, x._9 ) )
val dataframe = sqlContext.createDataFrame(rddSTG)

Answer 3

问题是-任何在scala中都是泛型类型。在您的情况下，NULL被视为ANY类型。

Spark根本不知道如何序列化NULL。

我们应该显式地提供一些特定的类型。

由于无法将null分配给Scala中的原始类型，因此您可以使用String来匹配列的其他值的数据类型。

所以试试这个：

case class Record(id: Int, name: String, score: Int, flag: String)
val sampleRdd = spark.sparkContext.parallelize(
  Seq(
    (1, null.asInstanceOf[String], 100, "YES"),
    (2, "RAKTOTPAL", 200, "NO"),
    (3, "BORDOLOI", 300, "YES"),
    (4, null.asInstanceOf[String], 400, "YES")))

sampleRdd.toDF("ID", "NAME", "SCORE","FLAG")

这样，df将保留空值。

用案例类

case class Record(id: Int, name: String, score: Int, flag: String)

val sampleRdd = spark.sparkContext.parallelize(
  Seq(
    Record(1, null.asInstanceOf[String], 100, "YES"),
    Record(2, "RAKTOTPAL", 200, "NO"),
    Record(3, "BORDOLOI", 300, "YES"),
    Record(4, null.asInstanceOf[String], 400, "YES")))
sampleRdd.toDF()