ZBLOG

Collections.sort(dataList); 这里报错了

Collections.sort(dataList); 报错的原因可能是因为 Tuple2<String, Long> 不是自然可排序的类型,或者你需要定义一个比较器来指定如何排序。由于 Tuple2 可能没有实现 Comparable 接口,因此你需要提供一个自定义的比较逻辑。

下面是修改后的代码示例,使用了自定义比较器来对 Tuple2<String, Long> 按照第二个值进行排序:

  1. import org.apache.flink.api.common.functions.MapFunction;
  2. import org.apache.flink.api.common.functions.ReduceFunction;
  3. import org.apache.flink.streaming.api.datastream.DataStream;
  4. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  5. import org.apache.flink.util.Collector;
  6. import scala.Tuple2;
  7. import java.util.ArrayList;
  8. import java.util.Collections;
  9. import java.util.Comparator;
  10. import java.util.List;
  11. public class YourFlinkJob {
  12. public static void main(String[] args) throws Exception {
  13. StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  14. // 假设你的输入数据为 DataStream<Tuple2<String, Long>>
  15. DataStream<Tuple2<String, Long>> inputData = ...; // 获取输入数据
  16. // 进行 keyBy 和 reduce 操作
  17. DataStream<Tuple2<String, Long>> reducedData = inputData
  18. .keyBy(k -> k.f0)
  19. .reduce(new ReduceFunction<Tuple2<String, Long>>() {
  20. @Override
  21. public Tuple2<String, Long> reduce(Tuple2<String, Long> value1, Tuple2<String, Long> value2) throws Exception {
  22. return Tuple2.of(value1.f0,value1.f1 + value2.f1);
  23. }
  24. });
  25. // 在这里处理和获取 top 500000 的元素
  26. DataStream<Tuple2<String, Long>> top500kData = reducedData
  27. .process(new TopNProcessFunction(500000)); // 自定义 Top N 逻辑
  28. // 输出 top 500k 到另一个文件中
  29. top500kData.writeAsText("output/top500k.txt", FileSystem.WriteMode.OVERWRITE);
  30. // 最终输出 reduced 数据到另一个文件中
  31. reducedData.writeAsText("output/reduced_data.txt", FileSystem.WriteMode.OVERWRITE);
  32. env.execute("Your Flink Job");
  33. }
  34. // 自定义 Process Function 实现 Top N 功能
  35. public static class TopNProcessFunction extends ProcessFunction<Tuple2<String, Long>, Tuple2<String, Long>> {
  36. private final int n;
  37. public TopNProcessFunction(int n) {
  38. this.n = n;
  39. }
  40. private List<Tuple2<String, Long>> dataList = new ArrayList<>();
  41. @Override
  42. public void processElement(Tuple2<String, Long> value, Context ctx,
  43. Collector<Tuple2<String, Long>> out) throws Exception {
  44. dataList.add(value);
  45. if (dataList.size() > n) {
  46. Collections.sort(dataList, new Comparator<Tuple2<String, Long>>() {
  47. @Override
  48. public int compare(Tuple2<String, Long> o1, Tuple2<String, Long> o2) {
  49. return o1.f1.compareTo(o2.f1); // 按照第二个值排序(从小到大)
  50. }
  51. });
  52. dataList.remove(dataList.size() - 1); // 移除最后一条(即最小的那个)
  53. }
  54. for (Tuple2 <String ,Long> item : dataList){
  55. out.collect(item);
  56. }
  57. }
  58. }
  59. }

修改说明:

  • 使用 Comparator 来指定如何根据 Tuple2 的第二个值 (f1) 排序。
  • 在维护列表时,当超过规定数量时,会进行排序并移除最后一条记录,从而只保留前 N 条记录。

这种方式在内存中维护一个固定大小的数据结构,可以有效地计算出前50万条记录。如果你的数据量非常大,请注意内存管理和状态后端配置,以确保作业能够顺利运行。


内容由零声教学AI助手提供,问题来源于学员提问

本站部分文章来源于网络,版权归原作者所有,如有侵权请联系站长删除。
转载请注明出处:https://golang.0voice.com/?id=21280

分享:
扫描分享到社交APP
上一篇
下一篇
发表列表
游客 游客
此处应有掌声~
评论列表

还没有评论,快来说点什么吧~

联系我们

在线咨询: 点击这里给我发消息

微信号:3007537140

上班时间: 10:30-22:30

关注我们
x

注册

已经有帐号?