reduce-on-pojo字段与apache-flink一起使用java

jum4pzuy 于 2021-06-25 发布在 Flink

关注(0)|答案(1)|浏览(351)

目前，我正在为一些分布式处理工具构建一个基准测试工具，在使用apache flink时遇到了一些问题。
设置很简单：logpojo是一个简单的pojo，有三个字段（长日期、双值、字符串数据）。在一个列表中，我正在寻找一个具有最小“value”字段的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;

所以它似乎无法应用reduce函数。我就是找不到，为什么。有什么提示吗？

Java reduce Aggregate Min apache-flink

来源：https://stackoverflow.com/questions/42689792/reduce-on-pojo-field-with-apache-flink-using-java

1条答案

按热度按时间

mbskvtky1#

首先你应该检查一下你的进口货。您从scala类中得到一个异常，但是您的程序是用java实现的。您可能意外地导入了scala数据集api。使用javaapi不应导致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;

赞(0）回复(0）举报 2021-06-25

我来回答

reduce-on-pojo字段与apache-flink一起使用java

1条答案

相关问题

热门标签

最新问答