我正在尝试使用RDD
创建DataFrame
。
首先,我使用下面的代码创建一个RDD
-
val account = sc.parallelize(Seq(
(1, null, 2,"F"),
(2, 2, 4, "F"),
(3, 3, 6, "N"),
(4,null,8,"F")))
It is working fine -
账户:org.apache.spark.rdd。RDD[(Int, Any, Int, String)] =ParallelCollectionRDD[0] at parallelize at:27
但是当尝试使用下面的代码
从RDD
创建DataFrame
时account.toDF("ACCT_ID", "M_CD", "C_CD","IND")
我正在得到低于错误
. lang。UnsupportedOperationException:类型Any的架构不是支持
我分析,每当我把null
的值在Seq
,然后只有我得到了错误。
是否有办法添加空值?
不使用rdd的替代方法:
import spark.implicits._
val df = spark.createDataFrame(Seq(
(1, None, 2, "F"),
(2, Some(2), 4, "F"),
(3, Some(3), 6, "N"),
(4, None, 8, "F")
)).toDF("ACCT_ID", "M_CD", "C_CD","IND")
df.show
+-------+----+----+---+
|ACCT_ID|M_CD|C_CD|IND|
+-------+----+----+---+
| 1|null| 2| F|
| 2| 2| 4| F|
| 3| 3| 6| N|
| 4|null| 8| F|
+-------+----+----+---+
df.printSchema
root
|-- ACCT_ID: integer (nullable = false)
|-- M_CD: integer (nullable = true)
|-- C_CD: integer (nullable = false)
|-- IND: string (nullable = true)
问题是Any是太一般的类型,而Spark不知道如何序列化它。您应该显式地提供一些特定的类型,在本例中为Integer
。因为在Scala中不能将null赋值给基本类型,所以你可以使用java.lang.Integer
。所以试试这个:
val account = sc.parallelize(Seq(
(1, null.asInstanceOf[Integer], 2,"F"),
(2, new Integer(2), 4, "F"),
(3, new Integer(3), 6, "N"),
(4, null.asInstanceOf[Integer],8,"F")))
输出:
rdd: org.apache.spark.rdd.RDD[(Int, Integer, Int, String)] = ParallelCollectionRDD[0] at parallelize at <console>:24
和相应的DataFrame:
scala> val df = rdd.toDF("ACCT_ID", "M_CD", "C_CD","IND")
df: org.apache.spark.sql.DataFrame = [ACCT_ID: int, M_CD: int ... 2 more fields]
scala> df.show
+-------+----+----+---+
|ACCT_ID|M_CD|C_CD|IND|
+-------+----+----+---+
| 1|null| 2| F|
| 2| 2| 4| F|
| 3| 3| 6| N|
| 4|null| 8| F|
+-------+----+----+---+
也可以考虑用更简洁的方式来声明空整数值,比如:
object Constants {
val NullInteger: java.lang.Integer = null
}