spark-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Ankur Gupta <ankur.gu...@cloudera.com.INVALID>
Subject Re: [DISCUSS] Enable blacklisting feature by default in 3.0
Date Mon, 01 Apr 2019 16:04:34 GMT
Hi Chris,

Thanks for sending over the example. As far as I can understand, it seems
that this would not have been a problem if
"spark.blacklist.application.maxFailedTasksPerExecutor" was set to a higher
threshold, as mentioned in my previous email.

Though, with 8/7 executors and 2 failedTasksPerExecutor, if the application
runs out of executors, that would imply at least 14 task failures in a
short period of time. So, I am not sure if the application should still
continue to run or fail. If this was not a transient issue, maybe failing
was the correct outcome, as it saves lot of unnecessary computation and
also alerts admins to look for transient/permanent hardware failures.

Please let me know if you think, we should enable blacklisting feature by
default with the higher threshold.

Thanks,
Ankur

On Fri, Mar 29, 2019 at 3:23 PM Chris Stevens <chris.stevens@databricks.com>
wrote:

> Hey All,
>
> My initial reply got lost, because I wasn't on the dev list. Hopefully
> this goes through.
>
> Back story for my experiments: customer was hitting network errors due to
> cloud infrastructure problems. Basically, executor X couldn't fetch from Y.
> The NIC backing the VM for executor Y was swallowing packets. I wanted to
> blacklist node Y.
>
> What I learned:
>
> 1. `spark.blacklist.application.fetchFailure.enabled` requires
> `spark.blacklist.enabled` to also be enabled (BlacklistTracker isn't
> created
> <https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala#L948>
without
> the latter). This was a problem because the defaults for
> `spark.blacklist.[task|stage|application].*` are aggressive and don't even
> apply to fetch failures. Those are always treated as non-transient. It
> would be nice to have fetch blacklisting without regular blacklisting.
>
> 2. Due to the conf coupling in #1 and transient cloud storage errors in
> the job (FileScanRDD was failing due to corrupted files), I had to set the
> `max*PerExecutor` and `max*PerNode` to really high values (i.e. 1000).
> Without these high settings, the customer was running out of nodes on the
> cluster (as we don't have blacklisting enabled by default, we haven't
> hooked it up to any sort of dynamic cloud VM re-provisioning - something
> like `killBlacklistedNodes`). Why? The same transient FileScanRDD failure
> hit over multiple stages, so even though executors were aggressively
> removed within one
> stage, `spark.blacklist.application.maxFailedTasksPerExecutor = 2` was
> reached. The stages were succeeding because the FileScanRDD attempts on
> other executors succeeded. As such, the 8 node cluster ran out of executors
> after 3 stages. I did not have `spark.blacklist.killBlacklistedExecutors`.
> If I did, then `spark.blacklist.application.maxFailedExecutorsPerNode`
> would have kicked in and the job might have failed after 4-6 stages,
> depending on how it played out. (FWIW, this was running one executor per
> node).
>
> -Chris
>
> On Fri, Mar 29, 2019 at 1:48 PM Ankur Gupta <ankur.gupta@cloudera.com>
> wrote:
>
>> Thanks Reynold! That is certainly useful to know.
>>
>> @Chris Will it be possible for you to send out those details if you still
>> have them or better create a JIRA, so someone can work on those
>> improvements. If there is already a JIRA, can you please provide a link to
>> the same.
>>
>> Additionally, if the concern is with the aggressiveness of the
>> blacklisting, then we can enable blacklisting feature by default with
>> higher thresholds for failures. Below is an alternate set of defaults that
>> were also proposed in the design document for max cluster utilization:
>>
>>    1. spark.blacklist.task.maxTaskAttemptsPerExecutor = 2
>>    2. spark.blacklist.task.maxTaskAttemptsPerNode = 2
>>    3. spark.blacklist.stage.maxFailedTasksPerExecutor = 5
>>    4. spark.blacklist.stage.maxFailedExecutorsPerNode = 4
>>    5. spark.blacklist.application.maxFailedTasksPerExecutor = 5
>>    6. spark.blacklist.application.maxFailedExecutorsPerNode = 4
>>    7. spark.blacklist.timeout = 5 mins
>>
>>
>>
>> On Fri, Mar 29, 2019 at 11:18 AM Reynold Xin <rxin@databricks.com> wrote:
>>
>>> We tried enabling blacklisting for some customers and in the cloud, very
>>> quickly they end up having 0 executors due to various transient errors. So
>>> unfortunately I think the current implementation is terrible for cloud
>>> deployments, and shouldn't be on by default. The heart of the issue is that
>>> the current implementation is not great at dealing with transient errors vs
>>> catastrophic errors.
>>>
>>> +Chris who was involved with those tests.
>>>
>>>
>>>
>>> On Thu, Mar 28, 2019 at 3:32 PM, Ankur Gupta <
>>> ankur.gupta@cloudera.com.invalid> wrote:
>>>
>>>> Hi all,
>>>>
>>>> This is a follow-on to my PR:
>>>> https://github.com/apache/spark/pull/24208, where I aimed to enable
>>>> blacklisting for fetch failure by default. From the comments, there is
>>>> interest in the community to enable overall blacklisting feature by
>>>> default. I have listed down 3 different things that we can do and would
>>>> like to gather feedback and see if anyone has objections with regards to
>>>> this. Otherwise, I will just create a PR for the same.
>>>>
>>>> 1. *Enable blacklisting feature by default*. The blacklisting feature
>>>> was added as part of SPARK-8425 and is available since 2.2.0. This feature
>>>> was deemed experimental and was disabled by default. The feature blacklists
>>>> an executor/node from running a particular task, any task in a particular
>>>> stage or all tasks in application based on number of failures. There are
>>>> various configurations available which control those thresholds.
>>>> Additionally, the executor/node is only blacklisted for a configurable time
>>>> period. The idea is to enable blacklisting feature with existing defaults,
>>>> which are following:
>>>>
>>>>    1. spark.blacklist.task.maxTaskAttemptsPerExecutor = 1
>>>>    2. spark.blacklist.task.maxTaskAttemptsPerNode = 2
>>>>    3. spark.blacklist.stage.maxFailedTasksPerExecutor = 2
>>>>    4. spark.blacklist.stage.maxFailedExecutorsPerNode = 2
>>>>    5. spark.blacklist.application.maxFailedTasksPerExecutor = 2
>>>>    6. spark.blacklist.application.maxFailedExecutorsPerNode = 2
>>>>    7. spark.blacklist.timeout = 1 hour
>>>>
>>>> 2. *Kill blacklisted executors/nodes by default*. This feature was
>>>> added as part of SPARK-16554 and is available since 2.2.0. This is a
>>>> follow-on feature to blacklisting, such that if an executor/node is
>>>> blacklisted for the application, then it also terminates all running tasks
>>>> on that executor for faster failure recovery.
>>>>
>>>> 3. *Remove legacy blacklisting timeout config*
>>>> : spark.scheduler.executorTaskBlacklistTime
>>>>
>>>> Thanks,
>>>> Ankur
>>>>
>>>
>>>

Mime
View raw message