我正在使用Spark的Structured Streaming(2.2.1)和Kafka一起工作,每60秒从传感器接收数据。我对如何打包这个Kafka数据以正确处理它感到困惑。
我需要在Kafka接收数据时进行一些计算。
我的问题是将从Kafka接收的JSON数据解压成我可以处理的数据集。
为了能够使用Spark,我已经为每个案例类结构创建了一些内容。
使用这个rawDataSet,我可以将每个单独的对象打包成数据集。
我需要在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
}
}
为了能够使用Spark,我已经为每个案例类结构创建了一些内容。
// 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 Schema
首先,我将从Kafka中获取的'value'字段放入我的通用schema中:
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对象的干净整洁的数据集。
处理数据
以下是我的问题,如果我想比较或计算设置和泵两个数据集之间的某些值,使用Structured Streaming时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;
我的做法是不正确的吗?还是有其他方法推荐来处理这个问题?
谢谢