D-Net

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

import com.googlecode.protobuf.format.JsonFormat;

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

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

import eu.dnetlib.data.mapreduce.hbase.propagation.communitytoresult.CommunityToResultFileReducer;

import eu.dnetlib.data.proto.ResultProtos;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

import java.util.HashSet;

import java.util.Iterator;

import java.util.Set;

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

public class PropagationCommunityThroughOrganizationFileReducer extends Reducer<ImmutableBytesWritable, Text, Text, Text> {

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

    private Text keyOut;

    private Text outValue;

    @Override

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

        super.setup(context);

        keyOut = new Text("");

        outValue = new Text();

    }

    @Override

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

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

        DedupedList communities = new DedupedList();

        Set<String> resultIds = new HashSet<>();

        while(it.hasNext()){

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

            switch (v.getType()){

                case fromorganization:

                    communities.addAll(DedupedList.fromJson(v.getValue()));

                    break;

                case fromresult:

                    resultIds.add(v.getValue());

                    break;

            }

        }

        if(communities.size() > 0){

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

            communities.stream().forEach(community->metadata.addContext(Utils.getContext(community, ORGANIZATION_COMMUNITY_TRUST, CLASS_ORGANIZATION_ID, DATA_INFO_TYPE,CLASS_ORGANIZATION_NAME)));

            for(String result: resultIds){

                keyOut.set(result);

                outValue.set(JsonFormat.printToString(Utils.getUpdate(metadata, result)).getBytes());

                context.write(keyOut, outValue);

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

            }

        }

    }

}