stratosphere/stratosphere — pull request #320 by andrehacker (2013-12-04 17:57:31+00:00)
Title: Accumulators

DON'T MERGE !!!

This pull request has the purpose to discuss the API for Accumulators (this is how I call it right now). I want to talk with Stephan tomorrow about implementation so good to have your feedback.

Please note: I mean Hadoop-style accumulators here: Write in the UDFs and ONLY read at the end of the job in the driver. Goal is also to make results available in next iteration.

I implemented the test case [```AccumulatorITCase```](https://github.com/andrehacker/ozone/blob/counters/pact/pact-tests/src/test/java/eu/stratosphere/pact/test/counters/AccumulatorITCase.java) which shows how you can use accumulators from the UDFs (see the Map UDF). I implemented 4 accumulators which I think should be shipped (see in the [Accumulators package](https://github.com/andrehacker/ozone/tree/counters/pact/pact-common/src/main/java/eu/stratosphere/pact/common/stubs/accumulators)). I think we should support convenience methods for these because it should be as simple as possible from UDF. The user can implement it's own accumulator type.

About the status: The counters are working locally in the UDFs, but are not merged and sent to the JobManager and Client (for clarity I don't write about the plans how to do this here). And we will probably have to move many classes around (e.g. Accumulator needs to be available in Nephele). But we should discuss about the API first.

@StephanEwen: I could not/didn't want to use the existing Aggregator classes for some reasons. Anyway, I think it's better to think of a good interface first and then see how we can merge these two.

#### Some design decisions
- Introduced ExecutionResult for local and remote executor to get counter results (not working yet, only partially implemented)
- Accumulators are identified by name (initially not by Enum as in Hadoop)
- All accumulators share a single namespace (i.e. stored in a single HashTable)
- No need to register accumulators, they are created on the fly in the UDFs. This means that type conflicts are possible, e.g. different UDFs use accumulators with same name but different type. I already check for this.

So far, please let me know what you think, e.g. about naming, functionality. We must be able to integrate this in Scala. Our interfaces Accumulator and SimpleAccumulator are comparable to Sparks [Accumulable](https://spark.incubator.apache.org/docs/0.6.2/api/core/spark/Accumulable.html) and [Accumulator](https://spark.incubator.apache.org/docs/0.6.2/api/core/spark/Accumulator.html) classes. They have a [test suite for accumulators](https://github.com/apache/incubator-spark/blob/2ce200bf7f7a38afbcacf3303ca2418e49bdbe2a/core/src/test/scala/org/apache/spark/AccumulatorSuite.scala) showing their functionality and API.