对 DStream 进行类型参数化答案

【问题标题】：Type-parameterize a DStream对 DStream 进行类型参数化
【发布时间】：2017-05-19 18:00:35
【问题描述】：

DStream 可以拥有type parameters 吗？

如果是，怎么做？

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

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

【问题讨论】：

标签： scala apache-spark streaming apache-kafka type-parameter

【解决方案1】：

Spark API 的大部分子集需要隐式 ClassTags（请参阅 Scala: What is a TypeTag and how do I use it?），PairDStreamFunctions.mapWithState 也不例外。检查class definition：

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

and:

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)

【讨论】：