数据流上的Apache-Flink随机离群点选择

Question

我试图使用ApacheFlink ML包的随机选择模型。

我不知道如何使用Kafka作为数据源，我知道它需要一个DataSet而不是DataStream，但我似乎无法窗口我的Kafka DataStream成为一个DataSet。

有没有一种方法可以将我的流视为一系列小数据集。例如，有没有办法说流中每10个元素匹配一个模式（按元素唯一ID滑动窗口），就将它们视为固定大小的数据集，并检测这个固定大小数据集中的任何异常值？

我想创建的场景是：

数据源-

我已经有了一个工作实现，直到预处理，我希望Flink能够满足我的要求？

Answer 1

我猜你可以创建一个基于计数的全局窗口，并使用执行环境来获取数据集。类似于以下内容可能会起作用（getResult将返回DataSet）：

      stream.
      keyBy(...).
      window(GlobalWindows.create).
      trigger(CountTrigger.of(10)).
      aggregate(new MyAggregator()).
      ...

    class MyAggregator extends AggregateFunction[..., ..., ...] {  

      var valueList: List[LabeledVector] = List[LabeledVector]()    

      override def createAccumulator(): MyAggregator = new MyAggregator()
      override def add(value: .., accumulator: MyAggregator): ... = ...
      override def merge(agg1: MyAggregator, agg2: MyAggregator): ... = ...
      override def getResult(accumulator: MyAggregator): ... = {
        ExecutionEnvironment.getExecutionEnvironment.fromCollection(valueList)
      }
    }