D-Net

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

import java.io.IOException;

import java.util.List;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

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 eu.dnetlib.data.mapreduce.hbase.Reporter;

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

	private String setName;

	private Agent agent;

	private Text keyout;

	private Text valueOut;

	private JsonParser parser;

	private ActionFactory factory;

	@Override

	protected void setup(final 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();

	}

	@Override

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

		final String inputJson = value.toString();

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

		try {

			context.getCounter(this.getClass().getSimpleName(), "total").increment(1);

			final List<AtomicAction> atomicActions =

					OrcidToActions.generatePublicationActionsFromDump(rootElement, factory, setName, agent,

							(Reporter) (counterGroup, counterName, delta) -> context.getCounter(counterGroup, counterName).increment(delta));

			if (atomicActions != null) {

				for (final AtomicAction action : atomicActions) {

					keyout.set(action.getRowKey());

					valueOut.set(action.toJSON());

					context.write(keyout, valueOut);

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

				}

				context.getCounter(this.getClass().getSimpleName(), "generated").increment(1);

			} else {

				context.getCounter(this.getClass().getSimpleName(), "skipped").increment(1);

			}

		} catch (final Throwable e) {

			System.err.println(inputJson);

			throw e;

		}

	}

}