在 Spark Structured Streaming 中处理二进制数据答案

【问题标题】：Process binary data in Spark Structured Streaming在 Spark Structured Streaming 中处理二进制数据
【发布时间】：2017-02-24 07:30:21
【问题描述】：

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

{"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 等等。如何检查结构化流中先前处理的数据集的状态？

【问题讨论】：

到目前为止你尝试了什么？
我已经更新了我的代码。请检查。我正在做 orderBy 然后 forEach 来处理数据。我被困在 processData 方法如何处理流接收的 Dataset 中的当前和以前的数据。

标签： java apache-spark spark-streaming apache-spark-dataset

【解决方案1】：

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

【讨论】：