类型参数化 DStream

Question

DStream 可以有 type parameter 吗？

如果是，怎么样？

当我在 myDStream: DStream[(A, B)]（class 参数）上尝试 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)

Answer 1

Spark API 的大量子集需要隐式 ClassTags（参见 Scala: What is a TypeTag and how do I use it?) and PairDStreamFunctions.mapWithState is no different. Check 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)

类型参数化 DStream

Type-parameterize a DStream

scala

apache-spark

streaming

apache-kafka

type-parameter