LpSolver extends BaseSolver implements Solver { @Override public final RLESparseResourceAllocation solve( final Map<RecurrenceId, List<ResourceSkyline>> jobHistory) throws SolverException, SkylineStoreException { preprocessor.validate(jobHistory, timeInterval); final List<ResourceSkyline> resourceSkylines = preprocessor.aggregateSkylines(jobHistory, minJobRuns); final int numJobs = resourceSkylines.size(); final int jobLen = getJobLen(resourceSkylines, numJobs); final ExpressionsBasedModel lpModel = new ExpressionsBasedModel(); Variable[] oa = new Variable[jobLen * numJobs]; Variable[] ua = new Variable[jobLen * numJobs]; Variable[] uaPredict = new Variable[jobLen * numJobs]; Variable[] x = new Variable[jobLen]; for (int i = 0; i < jobLen * numJobs; i++) { oa[i] = new Variable("oa" + i).lower(BigDecimal.valueOf(0)); ua[i] = new Variable("ua" + i).lower(BigDecimal.valueOf(0)); uaPredict[i] = new Variable("uaPredict" + i).lower(BigDecimal.valueOf(0)); } for (int i = 0; i < jobLen; i++) { x[i] = new Variable("x").lower(BigDecimal.valueOf(0)); } lpModel.addVariables(x); lpModel.addVariables(oa); lpModel.addVariables(ua); lpModel.addVariables(uaPredict); Variable eps = new Variable("epsilon").lower(BigDecimal.valueOf(0)); lpModel.addVariable(eps); int indexJobITimeK = 0; double cJobI = 0; double cJobITimeK = 0; ResourceSkyline resourceSkyline; int[] containerNums; Expression regularizationConstraint = lpModel.addExpression("regularization"); regularizationConstraint.set(eps, -numJobs); regularizationConstraint.upper(BigDecimal.valueOf(0)); for (int indexJobI = 0; indexJobI < resourceSkylines.size(); indexJobI++) { resourceSkyline = resourceSkylines.get(indexJobI); containerNums = preprocessor .getDiscreteSkyline(resourceSkyline.getSkylineList(), timeInterval, resourceSkyline.getContainerSpec().getMemorySize(), jobLen); cJobI = 0; for (int i = 0; i < containerNums.length; i++) { cJobI = cJobI + containerNums[i]; } for (int timeK = 0; timeK < jobLen; timeK++) { indexJobITimeK = indexJobI * jobLen + timeK; cJobITimeK = containerNums[timeK]; regularizationConstraint .set(uaPredict[indexJobITimeK], 1 / cJobI); generateOverAllocationConstraints(lpModel, cJobITimeK, oa, x, indexJobITimeK, timeK); generateUnderAllocationConstraints(lpModel, cJobITimeK, uaPredict, ua, x, indexJobITimeK, timeK); } } Expression objective = lpModel.addExpression("objective"); generateObjective(objective, numJobs, jobLen, oa, ua, eps); final Result lpResult = lpModel.minimise(); final TreeMap<Long, Resource> treeMap = new TreeMap<>(); RLESparseResourceAllocation result = new RLESparseResourceAllocation(treeMap, new DefaultResourceCalculator()); ReservationInterval riAdd; Resource containerSpec = resourceSkylines.get(0).getContainerSpec(); String pipelineId = ((RecurrenceId) jobHistory.keySet().toArray()[0]).getPipelineId(); Resource resource; for (int indexTimeK = 0; indexTimeK < jobLen; indexTimeK++) { riAdd = new ReservationInterval(indexTimeK * timeInterval, (indexTimeK + 1) * timeInterval); resource = Resource.newInstance( containerSpec.getMemorySize() * (int) lpResult .doubleValue(indexTimeK), containerSpec.getVirtualCores() * (int) lpResult .doubleValue(indexTimeK)); result.addInterval(riAdd, resource); LOGGER.debug("time interval: {}, container: {}.", indexTimeK, lpResult.doubleValue(indexTimeK)); } predictionSkylineStore.addEstimation(pipelineId, result); return result; }  @Override final void init(final Configuration config,
      PredictionSkylineStore skylineStore); @Override final RLESparseResourceAllocation solve(
      final Map<RecurrenceId, List<ResourceSkyline>> jobHistory); @Override final void close();  }
@Test public void testSolve() throws IOException, SkylineStoreException, SolverException, ResourceEstimatorException, DataFieldNotFoundException, ParseException { parseLog("src/test/resources/lp/tpch_q12.txt"); RecurrenceId recurrenceId = new RecurrenceId("tpch_q12", "*"); final Map<RecurrenceId, List<ResourceSkyline>> jobHistory = skylineStore.getHistory(recurrenceId); solver = createSolver(); RLESparseResourceAllocation result = solver.solve(jobHistory); String file = "src/test/resources/lp/answer.txt"; Reader fileReader = new InputStreamReader(new FileInputStream(file), Charset.forName("UTF-8")); BufferedReader bufferedReader = new BufferedReader(fileReader); String line = bufferedReader.readLine(); Configuration config = new Configuration(); config.addResource(new org.apache.hadoop.fs.Path( ResourceEstimatorConfiguration.CONFIG_FILE)); int timeInterval = config.getInt(ResourceEstimatorConfiguration.TIME_INTERVAL_KEY, 5); final long containerMemAlloc = jobHistory.entrySet().iterator().next().getValue().get(0) .getContainerSpec().getMemorySize(); int count = 0; int numContainer = 0; while (line != null) { numContainer = (int) (result.getCapacityAtTime(count * timeInterval).getMemorySize() / containerMemAlloc); assertEquals(Integer.parseInt(line), numContainer, 0.1 * Integer.parseInt(line)); line = bufferedReader.readLine(); count++; } fileReader.close(); bufferedReader.close(); }