private DataPartitioner createDataPartitioner(PDataPartitionFormat dpf,PDataPartitioner dataPartitioner,ExecutionContext ec) throws DMLRuntimeException {
  DataPartitioner dp=null;
  int numReducers=ConfigurationManager.getNumReducers();
  int maxNumRed=InfrastructureAnalyzer.getRemoteParallelReduceTasks();
  if (InfrastructureAnalyzer.isYarnEnabled())   maxNumRed=(int)Math.max(maxNumRed,YarnClusterAnalyzer.getNumCores() / 2);
  int numRed=Math.min(numReducers,maxNumRed);
switch (dataPartitioner) {
case LOCAL:
    dp=new DataPartitionerLocal(dpf,-1,_numThreads);
  break;
case REMOTE_MR:
dp=new DataPartitionerRemoteMR(dpf,-1,_ID,numRed,_replicationDP,MAX_RETRYS_ON_ERROR,ALLOW_REUSE_MR_JVMS,false);
break;
case REMOTE_SPARK:
dp=new DataPartitionerRemoteSpark(dpf,-1,ec,numRed,false);
break;
default :
throw new DMLRuntimeException("Undefined data partitioner: '" + dataPartitioner.toString() + "'.");
}
return dp;
}

private DataPartitioner createDataPartitioner ( PDataPartitionFormat dpf , PDataPartitioner dataPartitioner , ExecutionContext ec ) throws DMLRuntimeException { DataPartitioner dp = null ; int numReducers = ConfigurationManager . getNumReducers ( ) ; int maxNumRed = InfrastructureAnalyzer . getRemoteParallelReduceTasks ( ) ; if ( InfrastructureAnalyzer . isYarnEnabled ( ) ) maxNumRed = ( int ) Math . max ( maxNumRed , YarnClusterAnalyzer . getNumCores ( ) / 2 ) ; int numRed = Math . min ( numReducers , maxNumRed ) ; switch ( dataPartitioner ) { case LOCAL : dp = new DataPartitionerLocal ( dpf , - 1 , _numThreads ) ; break ; case REMOTE_MR : dp = new DataPartitionerRemoteMR ( dpf , - 1 , _ID , numRed , _replicationDP , MAX_RETRYS_ON_ERROR , ALLOW_REUSE_MR_JVMS , false ) ; break ; case REMOTE_SPARK : dp = new DataPartitionerRemoteSpark ( dpf , - 1 , ec , numRed , false ) ; break ; default : throw new DMLRuntimeException ( "Undefined data partitioner: '" + dataPartitioner . toString ( ) + "'." ) ; } return dp ; }
creates a new data partitioner according to the specified runtime parameter .
