[SPARK-8857][SPARK-8859][Core]Add an internal flag to Accumulable and send internal accumulator updates to the driver via heartbeats

[zsxwing]

This PR includes the following changes:

1. Remove the thread local Accumulators.localAccums. Instead, all Accumulators in the executors will register with its TaskContext.
2. Add an internal flag to Accumulable. For internal Accumulators, their updates will be sent to the driver via heartbeats.

[JoshRosen]

What happens if my accumulator update is huge? Could this lead to dropped heartbeats?

[JoshRosen]

Ah, I see now that it's only for specific accumulators and is internal-only.

[SparkQA]

Test build #37509 has finished for PR 7448 at commit 7182813.

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

[SparkQA]

Test build #37510 has finished for PR 7448 at commit bd7dcf1.

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

[rxin]

add internal to the scaladoc param to explain what it is

[rxin]

cc @andrewor14 for review also

[rxin]

I took a look at accumulator implementation -- one thing that is really annoying is boxing. Right now every update will box -- and it will really sucks for SQL if we do update per row.

In order for SQL to work better, I think we'd need to create a subtype of accumulator called LongAccumulator and DoubleAccumulator, that includes specialized methods for updating.

That can be done in a separate pull request.

[zsxwing]

Added comments.

In order for SQL to work better, I think we'd need to create a subtype of accumulator called LongAccumulator and DoubleAccumulator, that includes specialized methods for updating.

That can be done in a separate pull request.

Will do it in a follow up PR.

[rxin]

LGTM.

[SparkQA]

Test build #37559 has finished for PR 7448 at commit c24bc5b.

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

[rxin]

Thanks - I've merged this.

[andrewor14]

LGTM2, but it would be great to have some unit tests on this.