Here's a sample problem and a custom Accumulator solution in java, you could use this as a sample to your own use case. Input: HashMap<String, String> Output: HashMap<String, Int> that will contain the count for each key in the input HashMaps, Example: Input HashMaps: 1. {"key1", "Value1"}, {"key2", "Value2"} 2. {"key1", "Value2"} Output: {"key1", 2}, {"key2", 1} //Since key1 is repeated 2 times. Code: import org.apache.spark.util.AccumulatorV2;
import java.util.HashMap;
public class CustomAccumulator extends AccumulatorV2<HashMap<String, String>, HashMap<String, Integer>> {
private HashMap<String, Integer> outputHashMap;
public CustomAccumulator(){
this.outputHashMap = new HashMap<>();
}
@Override
public boolean isZero() {
return outputHashMap.size() == 0;
}
@Override
public AccumulatorV2<HashMap<String, String>, HashMap<String, Integer>> copy() {
CustomAccumulator customAccumulatorCopy = new CustomAccumulator();
customAccumulatorCopy.merge(this);
return customAccumulatorCopy;
}
@Override
public void reset() {
this.outputHashMap = new HashMap<>();
}
@Override
public void add(HashMap<String, String> v) {
v.forEach((key, value) -> {
this.outputHashMap.merge(key, 1, (oldValue, newValue) -> oldValue + newValue);
});
}
@Override
public void merge(AccumulatorV2<HashMap<String, String>, HashMap<String, Integer>> other) {
other.value().forEach((key, value) -> {
this.outputHashMap.merge(key, value, (oldValue, newValue) -> oldValue + newValue);
});
}
@Override
public HashMap<String, Integer> value() {
return this.outputHashMap;
}
}
... View more
