场景

现在有如下数据格式

图书分类，图书名，数量

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

场景解析

如果不基于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值，完成功能。

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

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