@Override public final RLESparseResourceAllocation solve(
      final Map<RecurrenceId, List<ResourceSkyline>> jobHistory)
      throws SolverException, SkylineStoreException {
    // TODO: addHistory timeout support for this function, and ideally we should
    // return the confidence
    // level associated with the predicted resource.
    preprocessor.validate(jobHistory, timeInterval);
    final List<ResourceSkyline> resourceSkylines =
        preprocessor.aggregateSkylines(jobHistory, minJobRuns);
    final int numJobs = resourceSkylines.size();
    final int jobLen = getJobLen(resourceSkylines, numJobs);

    /** Create variables. */
    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);

    /** Set constraints. */
    int indexJobITimeK = 0;
    double cJobI = 0;
    double cJobITimeK = 0;
    ResourceSkyline resourceSkyline;
    int[] containerNums;
    // 1. sum(job_i){sum(timeK){1/cJobI * uaPredict_job_i_timeK}} <= numJobs
    // * eps
    Expression regularizationConstraint =
        lpModel.addExpression("regularization");
    regularizationConstraint.set(eps, -numJobs);
    regularizationConstraint.upper(BigDecimal.valueOf(0)); // <= 0
    for (int indexJobI = 0;
         indexJobI < resourceSkylines.size(); indexJobI++) {
      resourceSkyline = resourceSkylines.get(indexJobI);
      // the # of containers consumed by job i in discretized time intervals
      containerNums = preprocessor
          .getDiscreteSkyline(resourceSkyline.getSkylineList(), timeInterval,
              resourceSkyline.getContainerSpec().getMemorySize(), jobLen);
      // the aggregated # of containers consumed by job i during its lifespan
      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;
        // the # of containers consumed by job i in the k-th time interval
        cJobITimeK = containerNums[timeK];
        regularizationConstraint
            .set(uaPredict[indexJobITimeK], 1 / cJobI);
        generateOverAllocationConstraints(lpModel, cJobITimeK, oa, x,
            indexJobITimeK, timeK);
        generateUnderAllocationConstraints(lpModel, cJobITimeK, uaPredict,
            ua, x, indexJobITimeK, timeK);
      }
    }

    /** Set objective. */
    Expression objective = lpModel.addExpression("objective");
    generateObjective(objective, numJobs, jobLen, oa, ua, eps);

    /** Solve the model. */
    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);

    /**
     * TODO: 1. We can calculate the estimated error (over-allocation,
     * under-allocation) of our prediction which could be used to generate
     * confidence level for our prediction; 2. Also, we can modify our model to
     * take job input data size (and maybe stage info) into consideration; 3. We
     * can also try to generate such conclusion: our prediction under-allocates
     * X amount of resources from time 0 to time 100 compared with 95% of
     * history runs; 4. We can build framework-specific versions of estimator
     * (such as scope/spark/hive, etc.) and provides more specific suggestions.
     * For example, we may say: for spark job i, its task size is X GB while the
     * container memory allocation is Y GB; as a result, its shuffling stage is
     * 20% slower than ideal case due to the disk spilling operations, etc. 5.
     * If we have more information of jobs (other than ResourceSkyline), we may
     * have such conclusion: job i is 20% slower than 90% of history runs, and
     * it is because part of its tasks are running together with job j's tasks.
     * In this case, we not only predict the amount of resource needed for job
     * i, but also how to place the resource requirements to clusters; 6. We may
     * monitor job progress, and dynamically increase/decrease container
     * allocations to satisfy job deadline while minimizing the cost; 7. We may
     * allow users to specify a budget (say $100 per job run), and optimize the
     * resource allocation under the budget constraints. 8. ...
     */
    return result;
  }
@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),
        StandardCharsets.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();
  }