结构化流式传输 mapGroupWithState 不适用于自定义接收器

Question

Spark 结构化流，尝试使用 mapgroupwithstate。有没有人遇到过 .format("console") 完美工作并完美打印增量状态更改的情况，但是每当我尝试更改 .format("anyStreamingSinkClass") 时，接收器类收到的数据帧只有当前批次但没有内存状态或增量效应。

case class WordCount(word:String,count:Int)
case class WordInfo(totalSum:Int)
case class WordUpdate(word:String,count:Int,expired:Boolean)


val ds = df.as[String].map{ x=>
  val arr = x.split(",",-1)
  WordCount( arr(0), arr(1).toInt )
}.groupByKey(_.word)
  .mapGroupsWithState[WordInfo,WordUpdate](GroupStateTimeout.NoTimeout()) {
  case( word:String, allWords:Iterator[WordCount], state:GroupState[WordInfo]) =>
    val events = allWords.toSeq
    val updatedSession = if (state.exists) {
      val existingState = state.get
      val updatedEvents = WordInfo(existingState.totalSum + events.map(event ⇒ event.count).sum)
      updatedEvents
    }
    else {
      WordInfo(events.map(event => event.count).sum)
    }
    state.update(updatedSession)

    WordUpdate(word,updatedSession.totalSum,false)

}


val query = ds  
  .writeStream
    //.format("console")
  .format("com.subhankar.streamDB.ConsoleSinkProvider")
  .outputMode(OutputMode.Update())
  .trigger(Trigger.ProcessingTime(3.seconds))
  //.option("truncate",false)
 .option("checkpointLocation","out.b")
  .queryName("q2090" )
  .start()

query.awaitTermination()

对于接收器格式，我得到 批次 21 的不同计数为 1 x,1 Batch 22 的不同计数为 1 x,2 批次 23 的不同计数为 1 x,3

对于我得到的控制台格式

-------------------------------------------
Batch: 1
-------------------------------------------
+----+-----+-------+
|word|count|expired|
+----+-----+-------+
|   x|    1|  false|
+----+-----+-------+

-------------------------------------------
Batch: 2
-------------------------------------------
+----+-----+-------+
|word|count|expired|
+----+-----+-------+
|   x|    3|  false|
+----+-----+-------+

-------------------------------------------
Batch: 3
-------------------------------------------
+----+-----+-------+
|word|count|expired|
+----+-----+-------+
|   x|    6|  false|
+----+-----+-------+

水槽做一个简单的打印...

override def addBatch(batchId: Long, data: DataFrame) = {

  val batchDistinctCount = data.rdd.distinct.count()
  if(data.count()>0) {
    println(s"Batch ${batchId}'s distinct count is ${batchDistinctCount}")
    println(data.map(x=> x.getString(0) + "," + x.getInt(1)).collect().mkString(","))
  }
}

Answer 1

我遇到了和你一样的问题。

当我在 Spark 2.2.0 上对其进行测试时，状态会在每个小批量之间重置并丢失。

然后我在Spark 2.3.0上测试了一下，结果变成了抛出异常：

Queries with streaming sources must be executed with writeStream.start()

通过这个异常，我发现我的客户 Sink 的操作不受支持。

在您的情况下，您不受支持的操作是 multiple aggregations。

你在一个小批量中有data.rdd.distinct.count() 和data.count() 和data.map，这就是所谓的多重聚合，并且被认为是不受支持的。

虽然在 Spark = 2.3 上它只会得到异常。

为了解决这个问题，以下避免多次聚合的修改可以获得正确的结果。

override def addBatch(batchId: Long, dataframe: DataFrame) = {
  val data = dataframe.collect()  // now do everything in this Array (care for OUT OF MEMORY)
  val batchDistinctCount = Set(data).size()
  if(data.length > 0) {
    println(s"Batch ${batchId}'s distinct count is ${batchDistinctCount}")
    println(data.map(x=> x.getString(0) + "," + x.getInt(1)).mkString(","))
  }
}