北京微网站建设设计服务域名查询网
背景
目前spark的repartition()方法是随机分配数据到下游，这会导致一个问题，有时候如果我们用repartition方法的时候，如果任务发生了重试，就有可能导致任务的数据不准确，那这个时候改怎么解决这个问题呢？
分析
在Spark RDD中存在着名为outputDeterministicLevel的变量，如下：
private[spark] final lazy val outputDeterministicLevel: DeterministicLevel.Value = {if (isReliablyCheckpointed) {DeterministicLevel.DETERMINATE} else {getOutputDeterministicLevel}}
那么该变量的作用是什么呢？让我们分析一下：
改变量最终会被Stage的isIndeterminate方法调用：
def isIndeterminate: Boolean = {rdd.outputDeterministicLevel == DeterministicLevel.INDETERMINATE}
而该方法会被DAGScheduler调用，有两处地方会被调用：
- submitMissingTasks中调用
private def submitMissingTasks(stage: Stage, jobId: Int): Unit = {logDebug("submitMissingTasks(" + stage + ")")// Before find missing partition, do the intermediate state clean work first.// The operation here can make sure for the partially completed intermediate stage,// `findMissingPartitions()` returns all partitions every time.stage match {case sms: ShuffleMapStage if stage.isIndeterminate && !sms.isAvailable =>mapOutputTracker.unregisterAllMapOutput(sms.shuffleDep.shuffleId)case _ =>}该方法主要用于在重新提交失败的stage时候，用来判断是否需要重新计算上游的所有任务。
- handleTaskCompletion中调用
case FetchFailed(bmAddress, shuffleId, _, mapIndex, _, failureMessage) =>。。。val noResubmitEnqueued = !failedStages.contains(failedStage)failedStages += failedStagefailedStages += mapStageif (noResubmitEnqueued) {// If the map stage is INDETERMINATE, which means the map tasks may return// different result when re-try, we need to re-try all the tasks of the failed// stage and its succeeding stages, because the input data will be changed after the// map tasks are re-tried.// Note that, if map stage is UNORDERED, we are fine. The shuffle partitioner is// guaranteed to be determinate, so the input data of the reducers will not change// even if the map tasks are re-tried.if (mapStage.isIndeterminate) {
这里如果任务Fetch失败了，根据该shuffle所对应的上游stage是不是isIndeterminate来向DAGScheduler提交ResubmitFailedStages事件，从而调用submitMissingTasks方法进行上游所有任务或者单个任务的重试。
再回到outputDeterministicLevel变量，该变量会调用getOutputDeterministicLevel方法进行循环调用上游的outputDeterministicLevel变量来确定outputDeterministicLevel的值。
结论
所以根据以上分析，我们可以改写对应的RDD的outputDeterministicLevel变量或者getOutputDeterministicLevel方法来进行stage任务的全部重试与否