D-Net

package eu.dnetlib.data.mapreduce.hbase.propagation.country.institutionalrepositories;

import java.io.IOException;

import java.util.List;

import eu.dnetlib.data.mapreduce.hbase.propagation.NotValidResultSequenceException;

import eu.dnetlib.data.mapreduce.hbase.propagation.ResultIterator;

import eu.dnetlib.data.mapreduce.hbase.propagation.Value;

import eu.dnetlib.data.mapreduce.util.OafRowKeyDecoder;

import eu.dnetlib.data.proto.OafProtos;

import org.apache.hadoop.hbase.io.ImmutableBytesWritable;

import org.apache.hadoop.hbase.mapreduce.TableReducer;

import org.apache.hadoop.hbase.util.Bytes;

import org.apache.hadoop.hbase.client.Put;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.hadoop.io.Text;

import static eu.dnetlib.data.mapreduce.hbase.propagation.PropagationConstants.*;

public class PropagationCountryFromDsOrgResultReducer extends TableReducer<InstOrgKey, Text, ImmutableBytesWritable>{

    private static final Log log = LogFactory.getLog(PropagationCountryFromDsOrgResultReducer.class);

    private ImmutableBytesWritable keyOut;

    @Override

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

        super.setup(context);

        keyOut = new ImmutableBytesWritable();

    }

    @Override

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

        ResultIterator rh = null;

        try {

            rh = new ResultCountryIterator(values,key.getKeyType().get());

        } catch (NotValidResultSequenceException e) {

            context.getCounter(COUNTER_PROPAGATION,e.getMessage()).increment(1);

            return;

        }

        context.getCounter(COUNTER_PROPAGATION,"institutional datasource").increment(1);

        while(rh.hasNext()){

            try{

                List<OafProtos.Oaf> oap = rh.next();

                byte[] oafUpdate = oap.get(0).toByteArray();

                byte[] targetRowKey = Bytes.toBytes(oap.get(0).getEntity().getId());

                OafRowKeyDecoder.decode(targetRowKey);

                final Put put = new Put(targetRowKey).add(Bytes.toBytes("result"), Bytes.toBytes("update_" + System.nanoTime()), oafUpdate);

                keyOut.set(targetRowKey);

                context.write(keyOut, put);

                context.getCounter(COUNTER_PROPAGATION, "added country to product").increment(1);

            }catch(IllegalArgumentException e){

                context.getCounter(COUNTER_PROPAGATION,"not valid result id in result list for country propagation").increment(1);

            }

        }

        if (!rh.getPropagate()){

            context.getCounter(COUNTER_PROPAGATION, "resultId expected in ordering was not found" ).increment(1);

        }

    }

}