我正在使用spark的结构化流媒体（2.2.1），使用kafka每60秒接收一次来自传感器的数据。我一直在思考如何 Package 这些Kafka数据，以便能够正确处理它。
我需要能够做一些计算，因为数据与Kafka进来。
我的问题是将来自kafka的json数据解包到我可以使用的数据集中

数据

简化的数据如下所示：

{
  id: 1,
  timestamp: "timestamp"
  pump: {
    current: 1.0,
    flow: 20.0
    torque: 5.0
  },
  reactors: [
    {
      id: 1,
      status: 200,
    },

    {
      id: 2,
      status: 300,
    }
  ],
  settings: {
    pumpTimer: 20.0,
    reactorStatusTimer: 200.0
  }
}

为了能够使用这个is spark，我为其中的每一个创建了一些case类结构：

// First, general package
case class RawData(id: String, timestamp: String, pump: String, reactors: Array[String], settings: String)
// Each of the objects from the data
case class Pump(current: Float, flow: Float, torque: Float)
case class Reactor(id: Int, status: Int)
case class Settings(oos: Boolean, pumpTimer: Float, reactorStatusTimer: Float)

以及使用以下方法生成模式：

val rawDataSchema = Encoders.product[RawData].schema

原始数据到spark架构

首先，我将Kafka的“值”字段放入我的通用模式中：

val rawDataSet = df.select($"value" cast "string" as "json")
  .select(from_json($"json", rawDataSchema))
  .select("data.*").as[RawData]

使用这个rawdataset，我可以将每个单独的对象打包成数据集。

val pump = rawDataSet.select(from_json($"pump", pumpSchema) as 'pumpData)
  .select("pumpData.*").as[Pump]

val settings = rawDataSet.select(from_json($"settings", settingsSchema) as 'settingsData)
  .select("settingsData.*").as[Settings]

这为每个json对象提供了漂亮干净的数据集。

使用数据

这里是我的问题，例如，如果我想比较或计算设置和泵的两个数据集之间的一些值，join不能使用结构化流。

val joinedData = pump.join(settings)

错误：

Exception in thread "main" org.apache.spark.sql.AnalysisException: Inner join between two streaming DataFrames/Datasets is not supported;

我对这一切的态度都错了吗？或者有没有其他处理方法的建议？
谢谢