unified architecture v0.1

app/com/linkedin/drelephant/AutoTuner.java

@@ -16,16 +16,11 @@
 
 package com.linkedin.drelephant;
 
+import com.linkedin.drelephant.tuning.Flow;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.log4j.Logger;
 
 import com.linkedin.drelephant.analysis.HDFSContext;
-import com.linkedin.drelephant.tuning.AzkabanJobCompleteDetector;
-import com.linkedin.drelephant.tuning.BaselineComputeUtil;
-import com.linkedin.drelephant.tuning.FitnessComputeUtil;
-import com.linkedin.drelephant.tuning.JobCompleteDetector;
-import com.linkedin.drelephant.tuning.PSOParamGenerator;
-import com.linkedin.drelephant.tuning.ParamGenerator;
 import com.linkedin.drelephant.util.Utils;
 
 import controllers.AutoTuningMetricsController;
@@ -43,20 +38,18 @@ public class AutoTuner implements Runnable {
   public static final long ONE_MIN = 60 * 1000;
   private static final Logger logger = Logger.getLogger(AutoTuner.class);
   private static final long DEFAULT_METRICS_COMPUTATION_INTERVAL = ONE_MIN / 5;
-
   public static final String AUTO_TUNING_DAEMON_WAIT_INTERVAL = "autotuning.daemon.wait.interval.ms";
+  private static final String tuningTypes[] = {"OBT"};
 
   public void run() {
 
     logger.info("Starting Auto Tuning thread");
     HDFSContext.load();
     Configuration configuration = ElephantContext.instance().getAutoTuningConf();
-
     Long interval =
         Utils.getNonNegativeLong(configuration, AUTO_TUNING_DAEMON_WAIT_INTERVAL, DEFAULT_METRICS_COMPUTATION_INTERVAL);
-
     try {
-      AutoTuningMetricsController.init();
+     /* AutoTuningMetricsController.init();
       BaselineComputeUtil baselineComputeUtil = new BaselineComputeUtil();
       FitnessComputeUtil fitnessComputeUtil = new FitnessComputeUtil();
       ParamGenerator paramGenerator = new PSOParamGenerator();
@@ -71,10 +64,22 @@ public void run() {
           logger.error("Error in auto tuner thread ", e);
         }
         Thread.sleep(interval);
+      }*/
+      AutoTuningMetricsController.init();
+      Flow autoTuningFlow= new Flow();
+      while (!Thread.currentThread().isInterrupted()) {
+        try {
+          autoTuningFlow.executeFlow();
+        } catch (Exception e) {
+          logger.error("Error in auto tuner thread ", e);
+        }
+        Thread.sleep(interval);
       }
     } catch (Exception e) {
       logger.error("Error in auto tuner thread ", e);
     }
     logger.info("Auto tuning thread shutting down");
   }
+
+
 }

app/com/linkedin/drelephant/tuning/AbstractAlgorithmManager.java

@@ -0,0 +1,44 @@
+package com.linkedin.drelephant.tuning;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.log4j.Logger;
+
+
+public abstract class AbstractAlgorithmManager implements Manager {
+  protected final String JSON_CURRENT_POPULATION_KEY = "current_population";
+  private final Logger logger = Logger.getLogger(getClass());
+  protected abstract List<JobTuningInfo> detectJobsForParameterGeneration();
+  protected ExecutionEngine _executionEngine;
+
+  protected  Boolean generateParameters(List<JobTuningInfo> jobsForParameterSuggestion){
+    List<JobTuningInfo> updatedJobTuningInfoList = new ArrayList<JobTuningInfo>();
+    for (JobTuningInfo jobTuningInfo : jobsForParameterSuggestion) {
+      JobTuningInfo newJobTuningInfo = generateParamSet(jobTuningInfo);
+      updatedJobTuningInfoList.add(newJobTuningInfo);
+    }
+    return true;
+  }
+
+
+  protected abstract JobTuningInfo generateParamSet(JobTuningInfo jobTuningInfo);
+
+
+  protected abstract Boolean updateDatabase(List<JobTuningInfo> tuningJobDefinitions);
+
+  public final Boolean execute() {
+    logger.info("Executing Tuning Algorithm");
+    Boolean parameterGenerationDone = false, databaseUpdateDone = false, updateMetricsDone = false;
+    List<JobTuningInfo> jobTuningInfo = detectJobsForParameterGeneration();
+    if (jobTuningInfo != null && jobTuningInfo.size() >= 1) {
+      logger.info("Generating Parameters ");
+      parameterGenerationDone = generateParameters(jobTuningInfo);
+    }
+    if (parameterGenerationDone) {
+      logger.info("Updating Database");
+      databaseUpdateDone = updateDatabase(jobTuningInfo);
+    }
+    logger.info("Param Generation Done");
+    return databaseUpdateDone;
+  }
+}

app/com/linkedin/drelephant/tuning/AbstractBaselineManager.java

@@ -0,0 +1,145 @@
+package com.linkedin.drelephant.tuning;
+
+import com.avaje.ebean.Ebean;
+import com.avaje.ebean.SqlRow;
+import com.linkedin.drelephant.mapreduce.heuristics.CommonConstantsHeuristic;
+import java.util.List;
+import models.TuningJobDefinition;
+import org.apache.commons.io.FileUtils;
+import org.apache.log4j.Logger;
+import controllers.AutoTuningMetricsController;
+
+
+public abstract class AbstractBaselineManager implements Manager {
+  protected final String BASELINE_EXECUTION_COUNT = "baseline.execution.count";
+  protected Integer NUM_JOBS_FOR_BASELINE_DEFAULT = 30;
+  protected String baseLineCalculationSQL =
+      "SELECT AVG(resource_used) AS resource_used, AVG(execution_time) AS execution_time FROM "
+          + "(SELECT job_exec_id, SUM(resource_used/(1024 * 3600)) AS resource_used, "
+          + "SUM((finish_time - start_time - total_delay)/(1000 * 60))  AS execution_time, "
+          + "MAX(start_time) AS start_time " + "FROM yarn_app_result WHERE job_def_id=:jobDefId "
+          + "GROUP BY job_exec_id " + "ORDER BY start_time DESC " + "LIMIT :num) temp";
+  protected String avgInputSizeSQL = "SELECT AVG(inputSizeInBytes) as avgInputSizeInMB FROM "
+      + "(SELECT job_exec_id, SUM(cast(value as decimal)) inputSizeInBytes, MAX(start_time) AS start_time "
+      + "FROM yarn_app_result yar INNER JOIN yarn_app_heuristic_result yahr " + "ON yar.id=yahr.yarn_app_result_id "
+      + "INNER JOIN yarn_app_heuristic_result_details yahrd " + "ON yahr.id=yahrd.yarn_app_heuristic_result_id "
+      + "WHERE job_def_id=:jobDefId AND yahr.heuristic_name='" + CommonConstantsHeuristic.MAPPER_SPEED + "' "
+      + "AND yahrd.name='" + CommonConstantsHeuristic.TOTAL_INPUT_SIZE_IN_MB + "' "
+      + "GROUP BY job_exec_id ORDER BY start_time DESC LIMIT :num ) temp";
+  private final Logger logger = Logger.getLogger(getClass());
+  protected Integer _numJobsForBaseline = null;
+
+  /*
+  Execute the whole logic for Base line Computing
+  1) Get the jobs for which Baseline have to be calculated
+  2) Calculate the baseline
+  3) Update the base line
+  4) Update the metrics .
+   */
+  @Override
+  public final Boolean execute() {
+    logger.info("Executing BaseLine");
+    Boolean baseLineComputationDone = false, databaseUpdateDone = false, updateMetricsDone = false;
+    List<TuningJobDefinition> tuningJobDefinitions = detectJobsForBaseLineComputation();
+    if (tuningJobDefinitions != null && tuningJobDefinitions.size() >= 1) {
+      logger.info("Computing BaseLine");
+      baseLineComputationDone = calculateBaseLine(tuningJobDefinitions);
+    }
+    if (baseLineComputationDone) {
+      logger.info("Updating Database");
+      databaseUpdateDone = updateDataBase(tuningJobDefinitions);
+    }
+    if (databaseUpdateDone) {
+      logger.info("Updating Metrics");
+      updateMetricsDone = updateMetrics(tuningJobDefinitions);
+    }
+    logger.info("Baseline Done");
+    return updateMetricsDone;
+  }
+
+  /**
+   * Fetches the jobs whose baseline is to be computed.
+   * This is done by returning the jobs with null average resource usage
+   * @return List of jobs whose baseline needs to be added
+   */
+
+  protected abstract List<TuningJobDefinition> detectJobsForBaseLineComputation();
+
+  /**
+   * Adds baseline metric values for a job
+   * @param tuningJobDefinitions Job for which baseline is to be computed
+   */
+
+  protected Boolean calculateBaseLine(List<TuningJobDefinition> tuningJobDefinitions) {
+    for (TuningJobDefinition tuningJobDefinition : tuningJobDefinitions) {
+      try {
+        logger.info("Computing and updating baseline metric values for job: " + tuningJobDefinition.job.jobName);
+        logger.debug("Running query for baseline computation " + baseLineCalculationSQL);
+        SqlRow baseline = Ebean.createSqlQuery(baseLineCalculationSQL)
+            .setParameter("jobDefId", tuningJobDefinition.job.jobDefId)
+            .setParameter("num", _numJobsForBaseline)
+            .findUnique();
+        Double avgResourceUsage = 0D;
+        Double avgExecutionTime = 0D;
+        avgResourceUsage = baseline.getDouble("resource_used");
+        avgExecutionTime = baseline.getDouble("execution_time");
+        tuningJobDefinition.averageExecutionTime = avgExecutionTime;
+        tuningJobDefinition.averageResourceUsage = avgResourceUsage;
+        tuningJobDefinition.averageInputSizeInBytes = getAvgInputSizeInBytes(tuningJobDefinition.job.jobDefId);
+        logger.debug(
+            "Baseline metric values: Average resource usage=" + avgResourceUsage + " and Average execution time="
+                + avgExecutionTime);
+      } catch (Exception e) {
+        logger.error("Error in computing baseline for job: " + tuningJobDefinition.job.jobName, e);
+        return false;
+      }
+    }
+    return true;
+  }
+
+  /**
+   * Returns the average input size in bytes of a job (over last _numJobsForBaseline executions)
+   * @param jobDefId job definition id of the job
+   * @return average input size in bytes as long
+   */
+  private Long getAvgInputSizeInBytes(String jobDefId) {
+    logger.debug("Running query for average input size computation " + avgInputSizeSQL);
+
+    SqlRow baseline = Ebean.createSqlQuery(avgInputSizeSQL)
+        .setParameter("jobDefId", jobDefId)
+        .setParameter("num", _numJobsForBaseline)
+        .findUnique();
+    Double avgInputSizeInBytes = baseline.getDouble("avgInputSizeInMB") * FileUtils.ONE_MB;
+    return avgInputSizeInBytes.longValue();
+  }
+
+  /**
+   * This method update database  for auto tuning monitoring for baseline computation
+   * @param tuningJobDefinitions
+   */
+  protected Boolean updateDataBase(List<TuningJobDefinition> tuningJobDefinitions) {
+    for (TuningJobDefinition tuningJobDefinition : tuningJobDefinitions) {
+      tuningJobDefinition.update();
+      logger.info("Updated baseline metric value for job: " + tuningJobDefinition.job.jobName);
+    }
+    return true;
+  }
+
+  /**
+   * This method update metrics for auto tuning monitoring for baseline computation
+   * @param tuningJobDefinitions
+   */
+
+  protected Boolean updateMetrics(List<TuningJobDefinition> tuningJobDefinitions) {
+    int baselineComputeWaitJobs = 0;
+    for (TuningJobDefinition tuningJobDefinition : tuningJobDefinitions) {
+      if (tuningJobDefinition.averageResourceUsage == null) {
+        baselineComputeWaitJobs++;
+      } else {
+        AutoTuningMetricsController.markBaselineComputed();
+      }
+    }
+    AutoTuningMetricsController.setBaselineComputeWaitJobs(baselineComputeWaitJobs);
+    return true;
+  }
+}