D-Net

package eu.dnetlib.data.mapreduce.hbase.dataimport;

import com.google.common.base.Joiner;

import com.google.gson.JsonObject;

import com.google.gson.JsonParser;

import eu.dnetlib.actionmanager.actions.ActionFactory;

import eu.dnetlib.actionmanager.actions.AtomicAction;

import eu.dnetlib.actionmanager.common.Agent;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

import java.util.List;

public class DOIBoostImportMapper extends Mapper<LongWritable, Text, Text, Text> {

    private String setName;

    private Agent agent;

    private Text keyout;

    private Text valueOut;

    private JsonParser parser;

    private ActionFactory factory;

    private boolean invisible;

    private boolean onlyOrganization;

    private static final String SEPARATOR = "@";

    @Override

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

        setName = context.getConfiguration().get("setName");

        agent = new Agent(context.getConfiguration().get("agentId"), context.getConfiguration().get("agentName"), Agent.AGENT_TYPE.service);

        keyout = new Text("");

        valueOut = new Text("");

        factory = new ActionFactory();

        parser = new JsonParser();

        invisible = context.getConfiguration().getBoolean("invisible", false);

        onlyOrganization= context.getConfiguration().getBoolean("onlyOrganization", false);

    }

    @Override

    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        final String inputJson = value.toString();

        final JsonObject rootElement = parser.parse(inputJson).getAsJsonObject();

        try {

            List<AtomicAction> atomicActions = DOIBoostToActions.generatePublicationActionsFromDump(rootElement, factory, setName, agent, invisible, onlyOrganization);

            if (atomicActions!= null) {

                for (AtomicAction action: atomicActions){

                    keyout.set(Joiner.on(SEPARATOR).join(action.getTargetRowKey(), action.getTargetColumnFamily(), action.getTargetColumn()));

                    valueOut.set(action.toJSON());

                    context.write(keyout, valueOut);

                    context.getCounter(this.getClass().getSimpleName(), action.getTargetColumnFamily()).increment(1);

                }

            }

        } catch (Throwable e) {

            System.err.println(inputJson);

            throw e;

        }

    }

}

package eu.dnetlib.data.mapreduce.hbase.dataimport;

import eu.dnetlib.actionmanager.actions.AtomicAction;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

import java.util.Iterator;

public class DOIBoostImportReducer extends Reducer<Text, Text, Text, Text> {

    private Text keyout;

    private Text valueOut;

    @Override

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

        keyout = new Text("");

        valueOut = new Text("");

    }

    @Override

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

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

        if (iterator.hasNext()) {

            final String aaString = iterator.next().toString();

            final AtomicAction aa = AtomicAction.fromJSON(aaString);

            keyout.set(aa.getRowKey());

            valueOut.set(aaString);

            context.write(keyout, valueOut);

            context.getCounter(this.getClass().getSimpleName(), aa.getTargetColumnFamily()).increment(1);

        }

        while (iterator.hasNext()) {

            final AtomicAction aa = AtomicAction.fromJSON(iterator.next().toString());

            context.getCounter(this.getClass().getSimpleName(), String.format("ignored %s", aa.getTargetColumnFamily())).increment(1);

        }

    }

}