D-Net

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

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

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

import eu.dnetlib.data.proto.*;

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.Iterator;

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

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

    private static final Log log = LogFactory.getLog(CommunityToResultReducer.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 {

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

        final ResultProtos.Result.Metadata.Builder metadata = ResultProtos.Result.Metadata.newBuilder();

        while(it.hasNext()){

            Value v = Value.fromJson(it.next().toString());

            metadata.addContext(Utils.getContext(v.getValue(),v.getTrust()));

        }

        final Put put = new Put(Bytes.toBytes(key.toString())).add(Bytes.toBytes("result"), Bytes.toBytes("update_" + System.nanoTime()), Utils.getUpdate(metadata, key.toString()).toByteArray());

        keyOut.set(Bytes.toBytes(key.toString()));

        context.write(keyOut, put);

        context.getCounter(COUNTER_PROPAGATION, "added community to result").increment(1);

    }

}