Spark 流式传输 - 自定义接收器和数据帧推断架构答案

【问题标题】：Spark streaming - Custom receiver and dataframe infer schemaSpark 流式传输 - 自定义接收器和数据帧推断架构
【发布时间】：2017-03-04 15:56:32
【问题描述】：

在接收器处考虑下面的代码 sn-p

val incomingMessage = subscriberSocket.recv(0)
val stringMessages = new String(incomingMessage).stripLineEnd.split(',')
store(Row.fromSeq(Array(stringMessages(0)) ++ stringMessages.drop(2)))

在接收方，我不想将表（由 stringMessages(0) 指示）每个列类型转换为实际的表类型。

在代码的主要部分，当我这样做时

val df = sqlContext.createDataFrame(eachGDNRdd,getSchemaAsStructField)
println(df.collect().length)

我收到以下错误

java.lang.ClassCastException: java.lang.String cannot be cast to java.lang.Double
        at scala.runtime.BoxesRunTime.unboxToDouble(BoxesRunTime.java:119)
        at org.apache.spark.sql.catalyst.expressions.BaseGenericInternalRow$class.getDouble(rows.scala:44)

现在，模式由 String 和 Int 字段组成。我已经交叉验证，该字段按类型匹配。但是，看起来 spark 数据框并没有推断出类型。

问题
1. spark不应该在运行时推断模式的类型（除非有矛盾）？
2. 由于表是动态的，架构会根据每行的第一个元素（包含表名）而有所不同。是否有任何简单的建议方法可以即时修改架构？

或者我错过了什么明显的东西？

【问题讨论】：

标签： scala apache-spark dataframe

【解决方案1】：

我是 Spark 新手，你没有说你运行的是什么版本，但是在 v2.1.0 中，由于你提到的特定原因，默认情况下模式推断是禁用的；如果记录结构不一致，Spark 就无法可靠地推断模式。您可以通过将 spark.sql.streaming.schemaInference 设置为 true 来启用架构推断，但我认为您最好自己指定架构。

【讨论】：