本文共 1281 字，大约阅读时间需要 4 分钟。

---

session聚合统计：

统计出来之前通过条件过滤的session，访问时长在0s~3s的session的数量，占总session数量的比例；4s-6s。。。。；

访问步长在1-3的session的数量，占总session数量的比例；4~6。。。；

Accumulator 1s_3s = sc.accumulator(0L);。。。。。。十几个Accumulator

传统实现方式

可以对过滤以后的session，调用foreach也可以，遍历所有session；计算每个session的访问时长和访问步长；

访问时长：把session的最后一个action的时间，减去第一个action的时间 访问步长：session的action数量 计算出访问时长和访问步长以后，根据对应的区间，找到对应的Accumulator，1s_3s.add(1L) 同时每遍历一个session，就可以给总session数量对应的Accumulator，加1 最后用各个区间的session数量，除以总session数量，就可以计算出各个区间的占比了

弊端

这种传统的实现方式，有什么不好？？？

最大的不好，就是Accumulator太多了，不便于维护

• 首先第一，很有可能，在写后面的累加代码的时候，比如找到了一个4s~6s的区间的session，但是却代码里面不小心，累加到7s-9s里面去了；
• 第二，当后期，项目如果要出现一些逻辑上的变更，比如说，session数量的计算逻辑，要改变，就得更改所有Accumulator对应的代码；或者说，又要增加几个范围，那么又要增加多个Accumulator，并且修改对应的累加代码；维护成本，相当之高（甚至可能，修改一个小功能，或者增加一个小功能，耗费的时间，比做一个新项目还要多；甚至于，还修改出了bug，那就耗费更多的时间）
  所以，我们这里的设计，不打算采用传统的方式，用十几个，甚至二十个Accumulator，因为维护成本太高

项目实现思路

这里的实现思路是，我们自己自定义一个Accumulator，实现较为复杂的计算逻辑，一个Accumulator维护了所有范围区间的数量的统计逻辑

• 低耦合，如果说，session数量计算逻辑要改变，那么不用变更session遍历的相关的代码；只要维护一个Accumulator里面的代码即可；
• 如果计算逻辑后期变更，或者加了几个范围，那么也很方便，不用多加好几个Accumulator，去修改大量的代码；只要维护一个Accumulator里面的代码即可；
  维护成本，大大降低

自定义Accumulator，也是Spark Core中，属于比较高端的一个技术

使用自定义Accumulator，大家就可以任意的实现自己的复杂分布式计算的逻辑 如果说，你的task，分布式，进行复杂计算逻辑，那么是很难实现的（借助于redis，维护中间状态，借助于zookeeper去实现分布式锁） 但是，使用自定义Accumulator，可以更方便进行中间状态的维护，而且不用担心并发和锁的问题

转载地址：https://blog.csdn.net/weixin_37850264/article/details/112257861 如侵犯您的版权，请留言回复原文章的地址，我们会给您删除此文章，给您带来不便请您谅解！