Hey,
I would like to propose a way to extend the standard Streaming Scala API
methods (map, flatmap, filter etc) with versions that take stateful
functions as lambdas. I think this would eliminate the awkwardness of
implementing RichFunctions in Scala and make statefulness more explicit:
*For example:*
def map( statefulMap: (I, Option[S]) => (O, Option[S]) )
def flatMap( statefulFlatMap: (I, Option[S] ) => (Traversable[O],
Option[S]))
This would be translated into RichMap and RichFlatMapFunctions that store
Option[S] as OperatorState for fault tolerance.
*Example rolling sum by key:*
val input: DataStream[Long] = ...
val sumByKey: DataStream[Long] =
input.keyBy(...).map( (next: Long, sum: Option[Long]) =>
sum match {
case Some(s) => (next + s, Some(next + s))
case None => (next, Some(next))
})
What do you think?
Gyula
