使用Java使用Apache Flink在Pojo字段上减少

Question

目前我正在为某些分布式处理工具构建基准测试工具，并且在Apache Flink上遇到了一些麻烦。

设置很简单：logpojo是一个简单的POJO，具有三个字段（长期，双值，字符串数据）。在列表中，我正在寻找具有最小"值"字段的一个logpojo。基本上等于：

pojoList.stream().min(new LogPojo.Comp()).get().getValue();

我的flink设置看起来像：

public double processLogs(List<LogPojo> logs) {
    final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
    DataSet<LogPojo> logSet = env.fromCollection(logs);
    double result = 0.0;
    try {
        ReduceOperator ro = logSet.reduce(new LogReducer());
        List<LogPojo> c = ro.collect();
        result = c.get(0).getValue();
    } catch (Exception ex) {
        System.out.println("Exception caught" + ex);
    }
    return result;
}
public class LogReducer implements ReduceFunction<LogPojo> {
    @Override
    public LogPojo reduce(LogPojo o1, LogPojo o2) {
        return (o1.getValue() < o2.getValue()) ? o1 : o2;
    }
}

它停止：

Exception in thread "main" java.lang.NoSuchMethodError: scala.collection.immutable.HashSet$.empty()Lscala/collection/immutable/HashSet;

因此，某种程度上似乎无法应用减少功能。我只是找不到，为什么。有提示吗？

Answer 1

首先，您应该检查导入。您会从Scala类中获得例外，但是您的程序已在Java中实现。您可能意外地导入了Scala数据集API。使用Java API不应导致Scala异常（除非您使用的是依赖Scala的类）。

无论如何，Flink具有min，max等的内置聚合方法

DataSet<LogPojo> logSet = env.fromCollection(logs);
// map LogPojo to a Tuple1<Double> 
//   (Flink's built-in aggregation functions work only on Tuple types)
DataSet<Tuple1<Double>> values = logSet.map(new MapFunction<LogPojo, Tuple1<Double>>() {
    @Override
    public Tuple1<Double> map(LogPojo l) throws Exception {
      return new Tuple1<>(l.value);
    }
  });
// fetch the min value (at position 0 in the Tuple)
List<Tuple1<Double>> c = values.min(0).collect();
// get the first field of the Tuple
Double minVal = c.get(0).f0;