19

我正在尝试创建一个DataFrameusing RDD.

首先，我正在RDD使用以下代码创建一个 -

val account = sc.parallelize(Seq(
                                 (1, null, 2,"F"), 
                                 (2, 2, 4, "F"),
                                 (3, 3, 6, "N"),
                                 (4,null,8,"F")))

它工作正常 -

帐户：org.apache.spark.rdd.RDD[(Int, Any, Int, String)] = ParallelCollectionRDD[0] at parallelize at :27

但是当尝试使用以下代码创建DataFrame时RDD

account.toDF("ACCT_ID", "M_CD", "C_CD","IND")

我得到以下错误

java.lang.UnsupportedOperationException：不支持 Any 类型的架构

我分析说，每当我投入null价值时，Seq只有我得到了错误。

有没有办法添加空值？

4

2 回答 2

27

不使用 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)
于 2017-06-13T16:09:12.397 回答
20

问题是 Any 类型太笼统，Spark 只是不知道如何序列化它。在您的情况下，您应该明确提供一些特定类型Integer。由于 null 不能分配给 Scala 中的原始类型，因此您可以使用它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
}
于 2016-09-13T07:49:40.370 回答