Added config option for stageFailure count and documentation

Author: Ilya Ganelin

core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala

@@ -97,7 +97,7 @@ class DAGScheduler(
   private[scheduler] val failedStages = new HashSet[Stage]
 
   // The maximum number of times to retry a stage before aborting
-  val maxStageFailures = 5
+  val maxStageFailures = sc.conf.getInt("spark.stage.maxFailures", 5)
 
   // To avoid cyclical stage failures (see SPARK-5945) we limit the number of times that a stage
   // may be retried. However, it only makes sense to limit the number of times that a stage fails

docs/configuration.md

@@ -1153,6 +1153,14 @@ Apart from these, the following properties are also available, and may be useful
     Should be greater than or equal to 1. Number of allowed retries = this value - 1.
   </td>
 </tr>
+<tr>
+  <td><code>spark.stage.maxFailures</code></td>
+  <td>5</td>
+  <td>
+    Number of individual stage failures before aborting the stage and not retrying it.
+    Should be greater than or equal to 1. Number of allowed retries = this value - 1.
+  </td>
+</tr>
 </table>
 
 #### Dynamic Allocation