D-Net

package eu.dnetlib.data.mapreduce.hbase.propagation.orcidthroughproducts;

import com.googlecode.protobuf.format.JsonFormat;

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

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

import eu.dnetlib.data.mapreduce.hbase.propagation.projecttoresult.PropagationProjectToResultReducer;

import eu.dnetlib.data.mapreduce.hbase.propagation.projecttoresult.ResultProjectIterator;

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

import eu.dnetlib.data.proto.OafProtos;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

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.io.Text;

import java.io.IOException;

import java.util.List;

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

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

public class PropagationOrcidToResultReducer  extends TableReducer<ImmutableBytesWritable, Text, ImmutableBytesWritable> {

    private static final Log log = LogFactory.getLog(PropagationOrcidToResultReducer.class); // NOPMD by marko on 11/24/08 5:02 PM

    private ImmutableBytesWritable keyOut;

    @Override

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

        super.setup(context);

        keyOut = new ImmutableBytesWritable();

    }

    @Override

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

        ResultIterator rh = null;

        try {

            rh = new ResultOrcidIterator(values, Bytes.toString(key.copyBytes()));

        } catch (NotValidResultSequenceException e) {

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

            return;

        }

        while (rh.hasNext()) {

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

            if(oaf_list != null){

                for (OafProtos.Oaf oaf : oaf_list) {

                    byte[] targetRowKey = Bytes.toBytes(oaf.getEntity().getId());

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

                    keyOut.set(targetRowKey);

                    context.write(keyOut, put);

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

                }

            }

        }

    }

}