D-Net

package eu.dnetlib.dhp.common.counter;

import org.apache.spark.AccumulableParam;

import scala.Tuple2;

/**

 * Spark {@link AccumulableParam} for tracking multiple counter values using {@link NamedCounters}.

 * 

 * @author madryk

 */

public class NamedCountersAccumulableParam implements AccumulableParam<NamedCounters, Tuple2<String,Long>> {

    private static final long serialVersionUID = 1L;

    //------------------------ LOGIC --------------------------

    /**

     * Increments {@link NamedCounters} counter with the name same as the first element of passed incrementValue tuple

     * by value defined in the second element of incrementValue tuple.

     */

    @Override

    public NamedCounters addAccumulator(NamedCounters counters, Tuple2<String, Long> incrementValue) {

        counters.increment(incrementValue._1, incrementValue._2);

        return counters;

    }

    /**

     * Merges two passed {@link NamedCounters}.

     */

    @Override

    public NamedCounters addInPlace(NamedCounters counters1, NamedCounters counters2) {

        for (String counterName2 : counters2.counterNames()) {

            counters1.increment(counterName2, counters2.currentValue(counterName2));

        }

        return counters1;

    }

    /**

     * Returns passed initialCounters value without any modifications.

     */

    @Override

    public NamedCounters zero(NamedCounters initialCounters) {

        return initialCounters;

    }

}