Spark Memory Streams 继续运行？

问题描述

我有一个简单的内存流查询，但它一直在运行。

这是我的写查询

  val imstream = new MemoryStream[SomeData](1,spark.sqlContext)

  val mV = imstream.toDS()
    .groupByKey(...)
    .flatMapGroupsWithState......//returning the values without any logic

  val sQ = mV
    .writeStream
    .format("memory")
    .queryName(queryName)
    .option("checkpointLocation","src/checkpoint")
    .start


  val currentOffset: Offset = imstream.addData(dataList)
  val nextOffset = imstream.addData(Seq(.....))

  sQ.processAllAvailable()
  imstream.commit(currentOffset.asInstanceOf[LongOffset])
  imstream.commit(nextOffset.asInstanceOf[LongOffset])

  sQ.awaitTermination(5000L)

我用它来测试，但它一直在运行。我也试过 sQ.stop() 但它没有任何区别？

我是否遗漏了有关停止内存流的内容？

[如果没有 flatMapGroupsWithState 似乎也能正常工作]

解决方法

暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！

如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@）