scala – Spark mapWithState API说明

Why is the new value an Option[T] type? As far as I’ve seen,it was
always defined for me,and since the method is supposed to be called
with a new state,I don’t really see the point why it could be
optional.

这是一个Option [T],因为如果使用StateSpec.timeout设置超时,例如：

StateSpec.function(spec _).timeout(Milliseconds(5000))

然后,一旦函数超时,传入的值将为None,State [T]上的isTimingOut方法将产生true.这是有道理的,因为状态的超时并不意味着已经为指定的键到达了一个新值,并且通常比使用T传递null更加安全(无论如何都不能用于原语),因为您期望用户安全地操作选项[T].

您可以在Sparks实现中看到：

// Get the timed out state records,call the mapping function on each and collect the
// data returned
if (removeTimedoutData && timeoutThresholdTime.isDefined) {
  newStateMap.getByTime(timeoutThresholdTime.get).foreach { case (key,state,_) =>
    wrappedState.wrapTimingOutState(state)
    val returned = mappingFunction(batchTime,key,None,wrappedState) // <-- This.
    mappedData ++= returned
    newStateMap.remove(key)
  }
}

What does the return value mean? I tried to find some pointers in the
documentations and source code,but none of them describe what it is
used for. Since I’m modifying the state of a key using state.remove()
and state.update(),why would I have to do the same with return
values?

返回值是一种沿着spark图传递中间状态的方法.例如,假设我想更新我的状态,但也在我的管道中使用中间数据执行某些操作,例如：

dStream
  .mapWithState(stateSpec)
  .map(optionIntermediateResult.map(_ * 2))
  .foreachRDD( /* other stuff */)

该返回值正是允许我继续操作所述数据的原因.如果您不关心中间结果并且只想要完整状态,则输出None是完全正常的.

编辑：

我写了一个blog post(跟随这个问题),试图对API进行深入的解释.