在Spark结构化流中处理二进制数据

Question

我正在使用Kafka和Spark结构化流式处理。我收到以下格式的卡巴消息。

{"deviceId":"001","sNo":1,"data":"aaaaa"} 
{"deviceId":"002","sNo":1,"data":"bbbbb"} 
{"deviceId":"001","sNo":2,"data":"ccccc"} 
{"deviceId":"002","sNo":2,"data":"ddddd"} 

我读它像下面。

Dataset<String> data = spark 
     .readStream() 
     .format("kafka") 
     .option("kafka.bootstrap.servers", bootstrapServers) 
     .option(subscribeType, topics) 
     .load() 
     .selectExpr("CAST(value AS STRING)") 
     .as(Encoders.STRING()); 
Dataset<DeviceData> ds = data.as(ExpressionEncoder.javaBean(DeviceData.class)).orderBy("deviceId","sNo"); 
ds.foreach(event -> 
     processData(event.getDeviceId(),event.getSNo(),event.getData().getBytes()) 
);} 

private void processData(String deviceId,int SNo, byte[] data) 
{ 
    //How to check previous processed Dataset??? 
} 

以我JSON消息 “数据” 是字节[]的字符串形式。我有一个要求，我需要按照“sNo”的顺序处理给定“deviceId”的二进制“数据”。因此，对于“deviceId”=“001”，我必须处理“sNo”= 1，然后“sNo”= 2等二进制数据。如何在结构化流式传输中检查之前处理过的数据集的状态？任何样本或链接都会有很大的帮助。我是Spark的新手，请和我一起裸照。谢谢。

Answer 1

如果您正在寻找像DStream.mapWithState这样的状态管理，那么它在结构化流式处理中还不受支持。工作正在进行中。请检查 https://issues.apache.org/jira/browse/SPARK-19067。