D-Net

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

import com.google.gson.Gson;

import eu.dnetlib.data.bulktag.CommunityConfiguration;

import eu.dnetlib.data.bulktag.CommunityConfigurationFactory;

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

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

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

import eu.dnetlib.data.proto.ResultProtos;

import org.apache.commons.lang.StringUtils;

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

import java.util.Iterator;

import java.util.Set;

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

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

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

    private ImmutableBytesWritable keyOut;

    private CommunityConfiguration cc;

    @Override

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

        super.setup(context);

        keyOut = new ImmutableBytesWritable();

        final String conf = context.getConfiguration().get("tagging.conf");

        if (StringUtils.isBlank(conf)) {

            throw new IllegalArgumentException("missing organization community map configuration");

        }

        System.out.println("conf = " + conf);

        cc = CommunityConfigurationFactory.fromJson(conf);

    }

    @Override

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

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

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

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

        while(it.hasNext()){

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

            if(v.getType() == PropagationConstants.Type.fromorganization){

                DedupedList dedup_list = new Gson().fromJson(v.getValue(),DedupedList.class);

                dedup_list.stream().forEach(org -> communities.addAll(cc.getCommunityForOrganizationValue(org)));

                communities.addAll(cc.getCommunityForOrganizationValue(Bytes.toString(key.get())));

            }

            if(v.getType() == PropagationConstants.Type.fromresult){

                resultIds.add(v.getValue());

            }

        }

        resultIds.stream().forEach(result -> {

            try {

                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)));

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

                keyOut.set(Bytes.toBytes(result));

                context.write(keyOut, put);

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

            } catch (IOException e) {

                e.printStackTrace();

            } catch (InterruptedException e) {

                e.printStackTrace();

            }

        });

    }

}