A more general and accurate implementation of MetricsAggregator for Spark

[ericsahit]

A more general and accurate implementation of MetricsAggregator for Spark which is compatible with Spark on YARN as well as executor dynamic allocation mode.

What's new:

1. Memory: Use actual executor memory request considering memoryOverhead instead of only spark.executor.memory.
2. Time: Use running time of all existed executor, not duration(sum of task executing time) of executors only including of those which are alive while ending of application.

I test this feature for three real Spark applications on our YARN cluster and compare Aggregate Memory Resource Allocation(Resource used in dr-elephant) metric of applications in YARN(showed when spark application finished in AM page), dr-original and and dr-optimized Below is result of one application for example:

• YARN: 410620516 MB-seconds
• dr-original: 1394513510.4 MB-seconds
• dr-optimized: 405405696 MB-seconds

Consider YARN showed as the standard value,

• deviation of dr-original：(410620516-1394513510.4)/410620516=239.61%
• deviation of dr-optimized：(410620516-405405696)/410620516=1.27%

Average deviation of this feature is 1.88%, as average deviation of dr-original is 280.86%.

This is our first PR, so there has also some TODO work about code style, code reasonable, and etc, so please help us review and feel free to give us advise.

Future work we plan:

• Add cpu-time resource usage for CPU are also can be scheduled and CPU resource even is bottleneck resource in our company.
• Catch the real physical memory resource usage and optimize the Resource wasted metric.

Thanks!

[akshayrai]

@ericsahit, thanks for the PR. I'll take a look at it and add my comments.

[ericsahit]

@akshayrai , OK. I have some login problem with gitter, Anyone please feel free to give me any suggestions.

[shankar37]

_storageMemwastageBuffer should be configurable

[ericsahit]

I thought this variable will be useless in Resource aggregation of Spark, because like MapReduce, we should figure out way of identifying the actual physical memory usage of each Executor, then use (1-physicalMemoryUsage/ExecutorMemory) as resource wasted. But in our company we add feature of checking the actual physical memory usage embedded in our Spark branch(different from official spark version).
That's some suggestion about resource wasted metric calculation, and i will change _storageMemwastageBuffer for temporary.

[shankar37]

rewrite as "any difference after a buffer of 50% is wasted"

[shankar37]

Since you are returning when the task was resubmitted, you are using the last attempt's metrics not first attempt's metrics, right ?

[shankar37]

You are addion the duration of task's duration to executor's duration. I have a couple of questions here.

1. if there are multiple tasks running in parellel inside an executor wouldn't that count it same duration multiple times ?
2. the reverse case where the executor is idle. you are not counting that. shouldn't you be ?

[ericsahit]

1. Yes. Because in MapReduce, the basic resource allocation unit in resource manager(like YARN) will be one map or reduce task. But in Spark, the corresponding is Executor instead of computing task, because task is in form of thread and share memory and CPU resource of one Executor. We could see this difference like this picture Spark vs MapReduce.
2. As described below, although one Executor is idle, it also take CPU and memory logical resource in resource manager(YARN). So i suppose we should calculate this metric according to the lifecycle of Executor instead of Task.

[shankar37]

what's the magic number 384 ? also, can you create constants for each of these

[shankar37]

where are you using shuffleread, inputBytes etc. only startTime and finishTime are used, right ? Do you plan to use the rest later sometime

[cwsteinbach]

Hi @ericsahit @shankar37, is this PR still relevant or has it been superseded by one of the other Spark PRs?

[fusonghe]

/**
Controls the Compare Feature
*/
public static Result compare() {
DynamicForm form = Form.form().bindFromRequest(request());
String partialFlowExecId1 = form.get(COMPARE_FLOW_ID1);
partialFlowExecId1 = (partialFlowExecId1 != null) ? partialFlowExecId1.trim() : null;
String partialFlowExecId2 = form.get(COMPARE_FLOW_ID2);
partialFlowExecId2 = (partialFlowExecId2 != null) ? partialFlowExecId2.trim() : null;

List<AppResult> results1 = null;
List<AppResult> results2 = null;
if (partialFlowExecId1 != null && !partialFlowExecId1.isEmpty() && partialFlowExecId2 != null && !partialFlowExecId2.isEmpty()) {
  IdUrlPair flowExecIdPair1 = bestSchedulerInfoMatchGivenPartialId(partialFlowExecId1, AppResult.TABLE.FLOW_EXEC_ID);
  IdUrlPair flowExecIdPair2 = bestSchedulerInfoMatchGivenPartialId(partialFlowExecId2, AppResult.TABLE.FLOW_EXEC_ID);
  results1 = AppResult.find
      .select(AppResult.getSearchFields() + "," + AppResult.TABLE.JOB_DEF_ID + "," + AppResult.TABLE.JOB_DEF_URL
          + "," + AppResult.TABLE.FLOW_EXEC_ID + "," + AppResult.TABLE.FLOW_EXEC_URL)
      .where().eq(AppResult.TABLE.FLOW_EXEC_ID, flowExecIdPair1.getId()).setMaxRows(100)
      .fetch(AppResult.TABLE.APP_HEURISTIC_RESULTS, AppHeuristicResult.getSearchFields())
      .findList();
  results2 = AppResult.find
      .select(
          AppResult.getSearchFields() + "," + AppResult.TABLE.JOB_DEF_ID + "," + AppResult.TABLE.JOB_DEF_URL + ","
              + AppResult.TABLE.FLOW_EXEC_ID + "," + AppResult.TABLE.FLOW_EXEC_URL)
      .where().eq(AppResult.TABLE.FLOW_EXEC_ID, flowExecIdPair2.getId()).setMaxRows(100)
      .fetch(AppResult.TABLE.APP_HEURISTIC_RESULTS, AppHeuristicResult.getSearchFields())
      .findList();
}
return ok(comparePage.render(compareResults.render(compareFlows(results1, results2))));

}

how do conf azkaban scheduler.xml @ @ericsahit