spark--当分组遇到排序的解决思路

作者: 大数据首席数据师 | 来源:发表于2018-10-29 17:56 被阅读10次

    场景

    现在有如下数据格式

    图书分类,图书名,数量

    现在想统计全部分类中数量最多的书名以及数量

    场景解析

    如果不基于spark,我们来思考这个问题,数据量大内存是放不下,分类也不确定有多少类,图书名可能有重复,还需要合并计算。这种情况只能是分治,首先分类,把文件首先按照分类拆分成多个文件,每个文件中的数据都是图书名数量,然后根据图书名对数量进行合并,最后进行排序。

    spark思维转化

    上面的思路单独写这个程序没问题,但是如果基于spark就有点问题了,首先是分区的事情,想把数据准确落在不同的分区,且不重复,必须要先知道到底有多少分区。所以首先要统计分类种类,帮助以后分区。

    分区器

    //data是已经读取进来的图书分类的集合data.distinct().collect()

    有了数据就要应用分区器

    classMyPartionerextendsPartitioner{

        private Map<String, Integer> part = new HashMap<>();

        publicMyPartioner(List<String> data){

            int count = 0;

            for (String s : data) {

                part.put(s, count++);

            }

        }

        @Override    publicintnumPartitions(){

            return part.size();

        }

        @Override    publicintgetPartition(Object o){

            Keys info = (Keys) o;

            return part.get(((Keys) o).type);

        }

    }

    直接根据已经生成好的数据来进行分区。保证1个分类1个分区,这样就可以以后的部分就只关注排序即可。

    数据合并

    分区,分区器都准备好了,按照以前的思路,是不是应该把数据分散在不同的分区了。想法挺好,但是在分布式存储中,数据移动的成本很高,所以都是先对本地数据进行处理合并,减小数据量然后才进行数据的shuffle等分区操作,所以这里我们要做的其实是合并同类数据。

    textFile.mapToPair(lines ->newTuple2(name,count))                .reduceByKey((x, y)->x + y);

    这里是一个典型的单词计数的案例。

    接下来就是想着分区,然后排序,如果你查查api的话,你会发现并没用按照value排序的算子。如果要排序的话,一定是key。这里发生了一个冲突点,就是你是按照type分区,次数的type就是key,接下来排序,其实就是按照type来排。 发现了我们要依赖key完成两件事,一个是分区,一个是排序。分区靠type,排序靠count。这里的解决方案就是用对象。计数之后,得到的结果会是一个<bookname,totalcount>的tuple。这个明显无法继续下去了,你连分区的条件都没了。 bookname和type是一一对应的,所以这里合并统计的是bookname+type的结构体。这样就满足了分区的条件了。 为了把排序的因子给加上,我们做个map操作,把type和count组织成一个对象。

    classKeysimplementsSerializable{    String type;    Integer count;publicKeys(String type, Integer count){this.type = type;this.count = count;    }}

    这样就给了我们很大的空间,在分区器里,取出key来进行操作,在排序的时候,写一个比较器,按照count来进行排序。

    .repartitionAndSortWithinPartitions(newMyPartioner(collect),newKeyCompare());

    直接使用分区并且排序的算子帮我解决这个问题。

    小结

    在大数据环境下,数据的shuffle操作的代价很大,所以优先考虑合并数据,然后再进行分区等等。spark的算子大部分都是对key进行生效的,例如排序等等,对value的操作大部分是合并和迭代,并没有单独的排序出来。所以要合理利用java对象来组合key值,完成功能。

    大家喜欢多多关注,你的关注是我最大的动力。

    大家喜欢的可以关注我的微信公号号:首席数据师  你的关注是我最大的动力

    相关文章

      网友评论

      本文标题:spark--当分组遇到排序的解决思路

      本文链接:https://www.haomeiwen.com/subject/ilcitqtx.html