[SPARK-8582][Core]Optimize checkpointing to avoid computing an RDD twice

[zsxwing]

Unlike #7021, this PR uses an approach similar to persist to compute and checkpoint an RDD at the same job. When computing an RDD at the first time, each partition will be checkpointed and read back as an iterator (maybe compute it again, not sure which one is better). After finishing the job, we also check if all partitions are checkpointed, if not, we still need to launch a job to checkpoint the missing partitions.

[zsxwing]

I feel confused for synchronized in RDDCheckpointData: since RDD is not thread safe, why we need to make RDDCheckpointData thread safe?

[SparkQA]

Test build #44276 has finished for PR 9258 at commit c909ef0.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):\n * class CheckpointManager extends Logging\n

[ryan-williams]

just out of curiosity, any reason not to do an if/else here?

if (!isCheckpointedAndMaterialized && 
    checkpointData.exists(_.isInstanceOf[ReliableRDDCheckpointData[T]])) {
  SparkEnv.get.checkpointMananger.getOrCompute(
    this, checkpointData.get.asInstanceOf[ReliableRDDCheckpointData[T]], split, context)
} else {
  computeOrReadCache(split, context)
}

[tdas]

@andrewor14 Can you please take a look at this?

[SparkQA]

Test build #44355 has finished for PR 9258 at commit 8ae42e0.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):\n * class CheckpointManager extends Logging\n

[SparkQA]

Test build #44360 has finished for PR 9258 at commit d69f775.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #44417 has finished for PR 9258 at commit 824be91.

• This patch fails PySpark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[zsxwing]

retest this please

[zsxwing]

There is still one problem in this PR. Currently, if an RDD is not persisted, this RDD will be recomputed to generate an Iterator after checkpointing (We cannot use the original Iterator since it's consumed). However, recomputing RDD may be slower than reading from the checkpoint file if the RDD is very complicated. Since we don't know whether recomputing is faster than reading from the checkpoint file, maybe we should add an option to the checkpoint API and let the user make the decision.

[SparkQA]

Test build #44429 has finished for PR 9258 at commit 824be91.

• This patch fails PySpark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[andrewor14]

retest this please

[andrewor14]

@zsxwing I took a quick look and I have a high level question. Why not just do the checkpointing iterator? IIUC this approach involves reading the iterator back from disk to return the values. Wouldn't that be potentially expensive? Also, this doesn't fix it for local checkpointing.

If we have a general checkpointing iterator, then RDD doesn't have to change much and we don't need to introduce another CheckpointManager, which I find a little clunky.

[SparkQA]

Test build #44590 has finished for PR 9258 at commit 824be91.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[zsxwing]

@andrewor14 I just opened #9428 to take over #7021 and fix potential issues in the previous PR.