More on Spark Accumulators

The Power of Spark Accumulators

 Spark Accumulators, discussed in an earlier blog) are massively more powerful than Hadoop counters because they support multiple types of data.  I have not seem discussions of using accumulators holding large sets of data, something that some of the classes discussed here could certainly do. The code discussed here is available here.

The only things required for an accumulator are a an AccumulatorParam instance defining how to construct a zero element and how to combine multiple instances. 

AccumulatorParam use a Long as a Counter (accumulator)

/** * com.lordjoe.distributed.spark.LongAccumulableParam * User: Steve * Date: 11/12/2014 */ public class LongAccumulableParam implements AccumulatorParam<long>,Serializable { public static final LongAccumulableParam INSTANCE = new LongAccumulableParam(); private LongAccumulableParam() {} @Override public Long addAccumulator(final Long r, final Long t) { return r + t; } @Override public Long addInPlace(final Long r1, final Long r2) { return r1 + r2; } @Override public Long zero(final Long initialValue) { return 0L; } }

AccumulatorParam to accumulate a single string by concatenation

import org.apache.spark.*; import java.io.*; /** * com.lordjoe.distributed.spark.StringAccumulableParam * Usage * final Accumulator<string> myString = ctx.accumulator("", "Letter Statistics", new StringAccumulableParam("\t"); * ... * myString.add("xyxzzy"); * User: Steve * Date: 11/12/2014 */ public class StringAccumulableParam implements AccumulatorParam<string>,Serializable { private final String separator; public StringAccumulableParam(final String pSeparator) { separator = pSeparator; } public StringAccumulableParam( ) { this(","); } @Override public String addAccumulator(final String r, final String t) { if(r.isEmpty()) return t; if(t.isEmpty()) return r; return r + separator + t; } @Override public String addInPlace(final String r , final String t) { if(r.isEmpty()) return t; if(t.isEmpty()) return r; return r + separator + t; } @Override public String zero(final String initialValue) { return ""; } }

AccumulatorParam use a Set of Strings as an accumulator

import org.apache.spark.*; import java.io.*; import java.util.*; /** * com.lordjoe.distributed.spark.StringSetAccumulableParam * Accumulates a set of Strings * Usage * final Accumulator<Set<string>> myStringSet = ctx.accumulator(Collextions.emptySet(), "String Set", StringSetAccumulableParam.INSTANCE); * ... * Set<string> toAdd = new HashSet<string>(); * toAdd.add("xyxzzy") * myStringSet.add(toAdd); * User: Steve * Date: 11/12/2014 */ public class StringSetAccumulableParam implements AccumulatorParam<Set<string>>, Serializable { public static final StringSetAccumulableParam INSTANCE = new StringSetAccumulableParam(); private StringSetAccumulableParam() {} @Override public Set<string> addAccumulator(final Set<string> r, final Set<string> t) { r.addAll(t); // todo ask if we should make a new set return r; } @Override public Set<string> addInPlace(final Set<string> r, final Set<string> t) { r.addAll(t); return r; } @Override public Set<string> zero(final Set<string> initialValue) { return new HashSet<string>(initialValue); } }

How to use accumulators

Accumulators may be used in two ways. First, the accumulator may be created in code as a final variable in the scope of the function - this is especially useful for lambdas, functions created on line. 

The following is an illustration of this

Using an accumulator as a final local variable

// Make two accumulators using Statistics final Accumulator<long> totalLetters= ctx.accumulator(0L, "ttl", LongAccumulableParam.INSTANCE); JavaRDD<string> lines = ... JavaRDD<string> words = lines.flatMap(new FlatMapFunction<String, String>() { @Override public Iterable<string> call(final String s) throws Exception { // Handle accumulator here totalLetters.add((long)s.length()); // count all letters .... }); .... Long numberCalls = totalCounts.value();

Alternatively a function may be defined with an accumulator as a member variable. Here the function is defined as a class and later used. I prefer this approach to lambdas especially if significant work is done in the function.
In a later blog I will discuss using a base class for more sophisticated logging

Using an accumulator as a member variable

public class WordsMapFunctionWithCount implements FlatMapFunction<string string="">,Serializable { private final Accumulator<long> counts; public WordsMapFunctionWithCount(final Accumulator<long> pCounts) { counts = pCounts; } public Iterable<string> call(String s) { // keep count of letters counts.add(1L); ... other stuff } } // end class // main Accumulator<long> totalCounts= ctx.accumulator(0L, "ttl", LongAccumulableParam.INSTANCE); JavaRDD<string> lines = ... JavaRDD<string> words = lines.flatMap(new WordsMapFunctionWithCount()); ... Long numberCalls = totalCounts.value();