private DeepPartition [ ] calculateSplits ( DBCollection collection ) { BasicDBList splitData = getSplitData ( collection ) ; List < ServerAddress > serverAddressList = collection . getDB ( ) . getMongo ( ) . getServerAddressList ( ) ; if ( splitData == null ) { Pair < BasicDBList , List < ServerAddress > > pair = getSplitDataCollectionShardEnviroment ( getShards ( collection ) , collection . getDB ( ) . getName ( ) , collection . getName ( ) ) ; splitData = pair . left ; serverAddressList = pair . right ; } Object lastKey = null ; // Lower boundary of the first min split List < String > stringHosts = new ArrayList <> ( ) ; for ( ServerAddress serverAddress : serverAddressList ) { stringHosts . add ( serverAddress . toString ( ) ) ; } int i = 0 ; MongoPartition [ ] partitions = new MongoPartition [ splitData . size ( ) + 1 ] ; for ( Object aSplitData : splitData ) { BasicDBObject currentKey = ( BasicDBObject ) aSplitData ; Object currentO = currentKey . get ( MONGO_DEFAULT_ID ) ; partitions [ i ] = new MongoPartition ( mongoDeepJobConfig . getRddId ( ) , i , new DeepTokenRange ( lastKey , currentO , stringHosts ) , MONGO_DEFAULT_ID ) ; lastKey = currentO ; i ++ ; } QueryBuilder queryBuilder = QueryBuilder . start ( MONGO_DEFAULT_ID ) ; queryBuilder . greaterThanEquals ( lastKey ) ; partitions [ i ] = new MongoPartition ( 0 , i , new DeepTokenRange ( lastKey , null , stringHosts ) , MONGO_DEFAULT_ID ) ; return partitions ; }
Calculate splits .