类型-参数化DStream

Question

DStream可以有type parameter吗？

如果是，怎么做？

当我尝试lazy val qwe = mStream.mapWithState(stateSpec) on myDStream: DStream[(A, B)] (类参数)时，我得到：

value mapWithState is not a member of org.apache.spark.streaming.dstream.DStream[(A, B)]
    lazy val qwe = mStream.mapWithState(stateSpec)

Answer 1

Spark的实质性子集需要隐式ClassTags (参见Scala:什么是TypeTag，我如何使用它？)，而PairDStreamFunctions.mapWithState没有什么不同。检查类定义

class PairDStreamFunctions[K, V](self: DStream[(K, V)])
  (implicit kt: ClassTag[K], vt: ClassTag[V], ord: Ordering[K])

和

def mapWithState[StateType: ClassTag, MappedType: ClassTag](
    spec: StateSpec[K, V, StateType, MappedType]
  ): MapWithStateDStream[K, V, StateType, MappedType] = {
  ...
}

如果要创建一个在泛型对流上操作并使用mapWithState的函数，那么至少应该为KeyType和ValueType类型提供ClassTags：

def foo[T : ClassTag, U : ClassTag](
  stream: DStream[(T, U)], f: StateSpec[T, U, Int, Int]) = stream.mapWithState(f)

如果StateType和MappedType也是参数化的，那么您也需要对这些参数进行ClassTags：

def bar[T : ClassTag, U : ClassTag, V : ClassTag,  W : ClassTag](
  stream: DStream[(T, U)], f: StateSpec[T, U, V, W]) = stream.mapWithState(f)