最近开始做大数据，数据库mongodb,计算用的spark,加载数据用的mongo-spark connector官方连接器，mongo 在加载数据的时候也是懒加载，通过aggregate高性能管道去加载数据，提供了多个分区器用于加载数据，默认的是MongoSamplePartitioner，这些在官方文档上有的。
连接器使用的最新版2.2.0版本
https://docs.mongodb.com/spark-connector/current/

image.png

开始没有制定分区器，发现数据计算结果不准确.

        SparkSession session = SparkSession.builder().master("local").appName("MongoSparkJob." + appName + ".mrStage")
                .config(KEY_SPARK_MONGODB_INPUT_URI, "mongodb://" + srcMongoUri)
                .config(KEY_SPARK_MONGODB_OUTPUT_URI, "mongodb://" + dstMongoUri)
                .getOrCreate();
        long start = System.currentTimeMillis();

        // Create a JavaSparkContext using the SparkSession's SparkContext
        JavaSparkContext jsc = new JavaSparkContext(session.sparkContext());
        String startTime = yearString + monthString + "00000000";
        String endime = yearString + monthString + "99000000";
        Document filterCondition = new Document().append(filterDateField,new Document("$gt",startTime).append("$lt",endime));
        ReadConfig readConfig = ReadConfig.create(session);
        long srcCount = MongoConnector.apply(readConfig.asOptions()).withCollectionDo(readConfig, Document.class,
                new Function<MongoCollection<Document>, Long>() {
                    /**    */
                    private static final long serialVersionUID = 1L;
                    @Override
                    public Long call(MongoCollection<Document> v1) throws Exception {
                        return v1.count(filterCondition);
                    }
                });
        System.out.println("count in db ： "+srcCount);
/／／／／  readConfig = readConfig.withOption("partitioner", MySamplePartitioner.class.getName());
        JavaMongoRDD<Document> mongoRdd = MongoSpark.load(jsc, readConfig);
        Document filterDoc = new Document("$match",filterCondition);
        JavaMongoRDD<Document> filterRdd = mongoRdd.withPipeline(Collections.singletonList(filterDoc));
        org.apache.spark.util.LongAccumulator counter = session.sparkContext().longAccumulator();
        filterRdd.foreach(new VoidFunction<Document>() {
            @Override
            public void call(Document document) throws Exception {
                counter.add(1);
            }
        });
        System.out.println("count by partitioner： "+counter.value());

count in db ： 231560
count by partitioner： 222213

再往上查看原因，原来这个连接器加载数据就没有加载完全,而且每次加载的还不一样，然后一个个的分区器的试，只有MongoPaginateByCountPartitioner才加载了满足条件的全部数据，但是这个分区器，根据分区的个数，算出每个分区的条数numPerPartition后，再通过聚合命令 skip numPerPartition 来划定每个分区,数据量大的时候skip的性能很差，所以就自己研究源码一探究竟，所有分区器默认都是根据默认的_id分区，默认的分区器MongoSamplePartitioner加是通过聚合命令sort将_id 排序再用 sample命令 采样一定数量的_id,但是采样值没有包括满足条件中最小的_id 和最大的_id,因此两端的区间数据都没有加载出来才出现计算不准确，而且每次采样的也不一样才导致每次加载的数据量也有差异。

当然可以自己对MongoSamplePartitioner加以修改，再指定修改后分区器（配置写法如上段注释的代码），这样就加载正确了。后面才能愉快滴计算。

        val samples = connector.withCollectionDo(readConfig, {
            coll: MongoCollection[BsonDocument] =>
              coll.aggregate(List(
                Aggregates.`match`(matchQuery),
                Aggregates.sample(numberOfSamples),
                Aggregates.project(Projections.include(partitionKey)),
                Aggregates.sort(Sorts.ascending(partitionKey))
              ).asJava).allowDiskUse(true).into(new util.ArrayList[BsonDocument]()).asScala
          })
          val minKey = connector.withCollectionDo(readConfig, {
            coll: MongoCollection[BsonDocument] =>
              coll.aggregate(List(
                Aggregates.`match`(matchQuery),
                Aggregates.project(Projections.include(partitionKey)),
                Aggregates.sort(Sorts.ascending(partitionKey)),
                Aggregates.limit(1)
              ).asJava).allowDiskUse(true).into(new util.ArrayList[BsonDocument]()).asScala
          })
          val maxKey = connector.withCollectionDo(readConfig, {
            coll: MongoCollection[BsonDocument] =>
              coll.aggregate(List(
                Aggregates.`match`(matchQuery),
                Aggregates.project(Projections.include(partitionKey)),
                Aggregates.sort(Sorts.descending(partitionKey)),
                Aggregates.limit(1)
              ).asJava).allowDiskUse(true).into(new util.ArrayList[BsonDocument]()).asScala
          })
          def collectSplit(i: Int): Boolean = (i % samplesPerPartition == 0) || !matchQuery.isEmpty && i == count - 1
          val rightHandBoundaries = samples.zipWithIndex.collect {
            case (field, i)  if collectSplit(i)=> field.get(partitionKey)
          }
          val addMinMax = matchQuery.isEmpty
          rightHandBoundaries.insert(0,minKey.head.get(partitionKey))
          rightHandBoundaries.insert(rightHandBoundaries.size,maxKey.head.get(partitionKey))
          val partitions = PartitionerHelper.createPartitions(partitionKey, rightHandBoundaries, PartitionerHelper.locations(connector), addMinMax)
          if (!addMinMax) PartitionerHelper.setLastBoundaryToLessThanOrEqualTo(partitionKey, partitions)
          partitions
        }