如何在DataStream上迭代

Question

我是scala的新手。我有一个自定义类Analytics.scala，它有很少的变量(var a，var b，var c)。我在我的测试用例中得到了一个分析类型的DataStream，我想为每个对象设置var c的值为'0‘。

我尝试过在DataStream上使用map函数，但没有帮助。我还尝试将stream转换为list，然后遍历该list，但也不起作用。

流的类型为DataStreamAnalytics。这是我尝试过的：

stream.map(x => x.c=0)

val a = DataStreamUtils.collect(stream.javaStream).asScala.toArray.iterator
a.foreach(x => x.c=0)

在我的测试用例中，var c的值不会更改为0。

Answer 1

一般来说，Flink DataStream不是一个可以迭代一次就可以完成的有限集合--它是一个潜在的无限流，它只是不断地拥有更多的数据。

使用地图是正确的方法。但是，当您将贴图应用于流时，如

stream.map(x => x.c=0)

您正在描述流转换，而不是修改流本身。相反，您应该尝试

streamWhereCisZero = stream.map(x => x.c=0)

这将创建一个新的流，其中每个元素都将c设置为零。

Answer 2

这就是我迭代的方式。不确定这是否是最好的解决方案。

val collection = DataStreamUtils.collect(stream.javaStream)
val results: Seq[Analytics] = collection.asScala.toSeq
for (result <- results){
    result.c=0
}