我在 Scala 中有这个通用方法
def updateStateByKey[S](updateFunc: JFunction2[JList[V], Optional[S],
Optional[S]]) : JavaPairDStream[K, S] = { ... }
当我用 Java 调用它时,这两个都无法编译:
1
JavaPairDStream<String, Integer> stateDstream =
pairs.<Integer>updateStateByKey(...);
2
JavaPairDStream<String, Integer> stateDstream =
pairs.updateStateByKey(...);
如何正确调用该方法?
错误消息:
The method updateStateByKey(Function2<List<Integer>,Optional<S>,Optional<S>>,
int) in the type JavaPairDStream<String,Integer> is not applicable for
the arguments
(Function2<List<Integer>,Optional<Integer>,Optional<Integer>>,
HashPartitioner, JavaPairRDD<String,Integer>)
编辑: 整个函数调用(Java 8):
final Function2<List<Integer>, Optional<Integer>, Optional<Integer>> updateFunction =
(values, state) -> {
Integer newSum = state.or(0);
for (Integer value : values) {
newSum += value;
}
return Optional.of(newSum);
};
JavaPairDStream<String, Integer> stateDstream = pairs.updateStateByKey(
updateFunction
,
new HashPartitioner(context.defaultParallelism()), initialRDD);
编辑: 事实证明,泛型不是问题,而是参数与方法签名不匹配。
最佳答案
问题是您传入了一个 initialRDD
,而方法 updateStateByKey
没有它作为参数。
最接近的签名是:
updateStateByKey[S](updateFunc: Function2[List[V], Optional[S], Optional[S]],
partitioner: Partitioner): JavaPairDStream[K, S]
关于java - 在 Spark Streaming 中调用 updateStateByKey 时出错,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/29129195/