Spark Streaming MQTT - 在数据集上应用模式答案

【问题标题】：Spark Streaming MQTT - Apply schema on datasetSpark Streaming MQTT - 在数据集上应用模式
【发布时间】：2017-02-10 00:51:09
【问题描述】：

我正在研究 DataBricks (Spark 2.0.1-db1 (Scala 2.11))，我正在尝试使用 Spark Streaming 函数。我正在使用这些库：
- spark-sql-streaming-mqtt_2.11-2.1.0-SNAPSHOT.jar（见这里：http://bahir.apache.org/docs/spark/current/spark-sql-streaming-mqtt/）

以下命令给了我一个数据集：

val lines = spark.readStream  
      .format("org.apache.bahir.sql.streaming.mqtt.MQTTStreamSourceProvider")  
      .option("clientId", "sparkTest")  
      .option("brokerUrl", "tcp://xxx.xxx.xxx.xxx:xxx")  
      .option("topic", "/Name/data")  
      .option("localStorage", "dbfs:/models/mqttPersist")  
      .option("cleanSession", "true")  
      .load().as[(String, Timestamp)]

使用此 printSchema：

root  
 |-- value : string (nullable : true)  
 |-- timestamp : timestamp (nullable : true)

我想在我的数据集的“值”列上应用一个模式。你可以看到我的 json 架构如下。

root  
 |-- id : string (nullable = true)  
 |-- DateTime : timestamp (nullable = true)  
 |-- label : double (nullable = true)

是否可以直接在流中解析我的 json 以获得类似的东西：

root   
 |-- value : struct (nullable : true)  
   |-- id : string (nullable = true)  
   |-- DateTime : timestamp (nullable = true)  
   |-- label : double (nullable = true)  
 |-- timestamp : timestamp (nullable : true)

目前，我没有看到任何从 mqtt 解析 json 的方法，任何帮助都会非常好。

提前致谢。

【问题讨论】：

标签： json apache-spark streaming mqtt

【解决方案1】：

我今天也遇到了同样的问题！我使用 json4s 和 Jackson 来解析 json。

我如何获得流数据集（与您拥有的几乎相同）：

 val lines = spark.readStream
   .format("org.apache.bahir.sql.streaming.mqtt.MQTTStreamSourceProvider")
   .option("topic", topic)
   .option("brokerUrl",brokerUrl)
   .load().as[(String,Timestamp)]

我使用案例类定义了架构：

  case class  DeviceData(devicename: String, time: Long, metric: String, value: Long, unit: String)

使用 org.json4s.jackson.JsonMethods.parse 解析 JSON 列：

val ds = lines.map {
  row =>
    implicit val format = DefaultFormats
    parse(row._1).extract[DeviceData]
}

输出结果：

val query = ds.writeStream
  .format("console")
  .option("truncate", false)
  .start()

结果：

+----------+-------------+-----------+-----+----+
|devicename|time         |metric     |value|unit|
+----------+-------------+-----------+-----+----+
|dht11_4   |1486656575772|temperature|9    |C   |
|dht11_4   |1486656575772|humidity   |36   |%   |
+----------+-------------+-----------+-----+----+

我有点失望，我无法提出使用 Sparks 原生 json 解析的解决方案。相反，我们必须依靠杰克逊。如果您将文件作为流读取，则可以使用 spark 本机 json 解析。因此：

val lines = spark.readStream
  .....
  .json("./path/to/file").as[(String,Timestamp)]

但是对于 MQTT，我们不能这样做。

【讨论】：