Spark 有状态流随着时间的推移增加内存

【问题标题】：Spark stateful streaming increasing memory over timeSpark 有状态流随着时间的推移增加内存
【发布时间】：2018-09-03 03:24:58
【问题描述】：

我们使用 spark 结构化流式传输并使用 mapGroupWithState 在一段时间内聚合数据。随着超时发生，我们正在使用

删除状态

state.remove () 。

我们正在使用 ganglia 来监控集群的健康状况。堆内存在此期间不断增加。我怀疑即使在状态删除之后旧状态对象仍然在内存中并且从未清理过。调用 state.remove () 时对象会发生什么？提示表示赞赏。

【问题讨论】：

几个月前也有同样的问题，我尝试了很多都没有成功，我认为有内存泄漏或状态存储中的东西，你找到解决方法了吗？
我们启用了所有级别的日志记录。我们观察到的是状态移除功能延迟触发，仅在新事件发生时触发。内存清理也是渐进的。

标签： apache-spark garbage-collection spark-streaming

【解决方案1】：

我认为这应该是一个评论，但我刚刚创建了我的 Stack Overflow 帐户。

几周前，我在使用 Spark Streaming 时遇到了同样的问题。我的状态太大，我需要放弃旧状态。我使用 mapGroupWithState 并且它有效。起初我不明白 state.remove() 是如何工作的，但经过几次执行后我就明白了。

首先，什么时候调用 remove 方法？我使用了几分钟的超时时间。当我在时间到期后停止向该键发送消息时，我得到了超时和删除工作的方式。事件迭代器为空且 state.hasTimedOut 为真。

所以我决定在 if 中划分函数。

if (state.hasTimedOut) {
    state.remove()
    // Here I returned my object totally null
} else {
    // deal with new event
    // update state
    // set time out
}

也许您正在删除状态而不是跳过执行？当时间已过且键没有事件时，超时将为 true。

【讨论】：