commit e6c36168147d9c70c07c834acfaad575e08a5bb6
author Sanha Lee <sanhaleehana@gmail.com> Tue Aug 21 12:23:49 2018 +0900
committer Won Wook SONG <wonook@apache.org> Tue Aug 21 12:23:49 2018 +0900
tree 1196014de6b59acb16e32cb2cdb74e6c0f105fe5
parent faecf759ee21677c010250046d17e305238486a2

[NEMO-139, 6] Logic in the scheduler for appending jobs, Support RDD caching (#111)

JIRA: [NEMO-139: Logic in the scheduler for appending jobs](https://issues.apache.org/jira/projects/NEMO/issues/NEMO-139)
JIRA: [NEMO-6: Support RDD caching](https://issues.apache.org/jira/projects/NEMO/issues/NEMO-6)

**Major changes:**
- add a logic in the scheduler for appending plans (NEMO-139)
  - implement `PlanAppender` that appends submitted `PhysicalPlan` to a original `PhysicalPlan`
  - refactor `PlanStateManager`, `BatchScheduler`, `BlockManagerMaster`, and `TaskDispatcher` to reflect that all plans from a single job are appended to a single `PhysicalPlan` through `PlanAppender`
- support RDD caching (NEMO-6)
  - add `CacheIdProperty` property and `GhostProperty`
    - When a Spark user program call `cache()` or `persist()` for a `RDD`, the RDD creates a ghost vertex and connect the vertex having the `RDD` to the ghost vertex. This edge to the ghost vertex is annotated with an ID of cache (`cacheIdProperty`). When a plan with this edge is executed in our runtime, the data to cache will be stored in the edge as the required `StorageLevel` format. (Any extra feature is not required in our runtime to produce or sustain this data.)
    - When the `BatchScheduler` encounter a task that annotated with the `GhostProperty`, the vertex will not be scheduled but just regarded as a completed task.
  - implement `Optimizer` that conducts optimization by using `OptimizationPass`es from our `UserApplicationRunner` to separate the roll.
    - When an IR DAG that contains any edge with `cacheIdProperty` is submitted and there was any already executed IR DAG that contains an edge with the identical `cacheIdProperty`, the `Optimizer` crops the IR DAG before the cache edge and adds a `CachedSourceVertex` before the edge. 
  - make `PlanAppender` properly handle the caching
    - Make `PlanAppender` append the `PhysicalPlan` constructed from the cropped IR DAG with caching edge to the original `PhysicalPlan` and add a new edge from the vertex that has the actual  edge to a ghost vertex and the new `CachedSourceVertex`. In runtime, when the `CachedSourceVertex` requires the data, the cached data that produced and stored in the edge to the ghost vertex will be read through our `DuplicateEdgeGroupProperty` logic.

**Minor changes to note:**
- N/A.

**Tests for the changes:**
- add an integration test that tests `SparkCachingWordCount` application
    - `SparkCachingWordCount` caches a shuffle data and calculates that which keys have identical count by using the cached data.

**Other comments:**
- I'm sorry for the late. This issue is a part of our first release and the target due was August 16th, but it is delayed to resolve conflicts. 

Closes #111