Spark：以编程方式在 Scala 中创建数据框模式答案

【问题标题】：Spark: Programmatically creating dataframe schema in scalaSpark：以编程方式在 Scala 中创建数据框模式
【发布时间】：2016-07-18 22:31:58
【问题描述】：

我有一个小型数据集，它将是 Spark 作业的结果。为了在工作结束时方便起见，我正在考虑将此数据集转换为数据框，但一直在努力正确定义模式。问题是下面的最后一个字段（topValues）；它是元组的 ArrayBuffer —— 键和计数。

  val innerSchema =
    StructType(
      Array(
        StructField("value", StringType),
        StructField("count", LongType)
      )
    )
  val outputSchema =
    StructType(
      Array(
        StructField("name", StringType, nullable=false),
        StructField("index", IntegerType, nullable=false),
        StructField("count", LongType, nullable=false),
        StructField("empties", LongType, nullable=false),
        StructField("nulls", LongType, nullable=false),
        StructField("uniqueValues", LongType, nullable=false),
        StructField("mean", DoubleType),
        StructField("min", DoubleType),
        StructField("max", DoubleType),
        StructField("topValues", innerSchema)
      )
    )

  val result = stats.columnStats.map{ c =>
    Row(c._2.name, c._1, c._2.count, c._2.empties, c._2.nulls, c._2.uniqueValues, c._2.mean, c._2.min, c._2.max, c._2.topValues.topN)
  }

  val rdd = sc.parallelize(result.toSeq)

  val outputDf = sqlContext.createDataFrame(rdd, outputSchema)

  outputDf.show()

我得到的错误是 MatchError:scala.MatchError: ArrayBuffer((10,2), (20,3), (8,1)) (of class scala.collection.mutable.ArrayBuffer)

当我调试和检查我的对象时，我看到了这个：

rdd: ParallelCollectionRDD[2]
rdd.data: "ArrayBuffer" size = 2
rdd.data(0): [age,2,6,0,0,3,14.666666666666666,8.0,20.0,ArrayBuffer((10,2), (20,3), (8,1))]
rdd.data(1): [gender,3,6,0,0,2,0.0,0.0,0.0,ArrayBuffer((M,4), (F,2))]

在我看来，我已经准确地描述了我的 innerSchema 中元组的 ArrayBuffer，但 Spark 不同意。

知道我应该如何定义架构吗？

【问题讨论】：

如果您提供示例数据或至少提供rdd 的确切类型，这将很有用。

标签： scala apache-spark schema dataframe

【解决方案1】：

import spark.implicits._
import org.apache.spark.sql.types._
import org.apache.spark.sql.functions._


val searchPath = "/path/to/.csv"
val columns = "col1,col2,col3,col4,col5,col6,col7"
val fields = columns.split(",").map(fieldName => StructField(fieldName, StringType, 
nullable = true))
val customSchema = StructType(fields)
var dfPivot =spark.read.format("com.databricks.spark.csv").option("header","false").option("inferSchema", "false").schema(customSchema).load(searchPath)

与使用默认架构加载数据相比，使用自定义架构加载数据会更快

【讨论】：

【解决方案2】：

正如 David 所指出的，我需要使用 ArrayType。 Spark 对此很满意：

  val outputSchema =
    StructType(
      Array(
        StructField("name", StringType, nullable=false),
        StructField("index", IntegerType, nullable=false),
        StructField("count", LongType, nullable=false),
        StructField("empties", LongType, nullable=false),
        StructField("nulls", LongType, nullable=false),
        StructField("uniqueValues", LongType, nullable=false),
        StructField("mean", DoubleType),
        StructField("min", DoubleType),
        StructField("max", DoubleType),
        StructField("topValues", ArrayType(StructType(Array(
          StructField("value", StringType),
          StructField("count", LongType)
        ))))
      )
    )

【讨论】：

【解决方案3】：

val rdd = sc.parallelize(Array(Row(ArrayBuffer(1,2,3,4))))
val df = sqlContext.createDataFrame(
  rdd,
  StructType(Seq(StructField("arr", ArrayType(IntegerType, false), false)
)

df.printSchema
root
 |-- arr: array (nullable = false)
 |    |-- element: integer (containsNull = false)

df.show
+------------+
|         arr|
+------------+
|[1, 2, 3, 4]|
+------------+

【讨论】：

是的，ArrayType 是正确的方法。谢谢！我的最终模式在我的答案中。