D-Net

package eu.dnetlib.data.mapreduce.hbase.dataimport;

import eu.dnetlib.actionmanager.actions.AtomicAction;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

import java.util.Iterator;

public class DOIBoostImportReducer extends Reducer<Text, Text, Text, Text> {

    private Text keyout;

    private Text valueOut;

    @Override

    protected void setup(Context context) throws IOException, InterruptedException {

        keyout = new Text("");

        valueOut = new Text("");

    }

    @Override

    protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {

        final Iterator<Text> iterator = values.iterator();

        if (iterator.hasNext()) {

            final String aaString = iterator.next().toString();

            final AtomicAction aa = AtomicAction.fromJSON(aaString);

            keyout.set(aa.getRowKey());

            valueOut.set(aaString);

            context.write(keyout, valueOut);

            context.getCounter(this.getClass().getSimpleName(), aa.getTargetColumnFamily()).increment(1);

        }

        while (iterator.hasNext()) {

            final AtomicAction aa = AtomicAction.fromJSON(iterator.next().toString());

            context.getCounter(this.getClass().getSimpleName(), String.format("ignored %s", aa.getTargetColumnFamily())).increment(1);

        }

    }

}