D-Net

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

import java.io.IOException;

import org.apache.hadoop.hbase.client.Result;

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

import org.apache.hadoop.hbase.mapreduce.TableMapper;

import org.apache.hadoop.io.Text;

import eu.dnetlib.data.mapreduce.JobParams;

import eu.dnetlib.data.mapreduce.util.DedupUtils;

import eu.dnetlib.data.mapreduce.util.OafDecoder;

import eu.dnetlib.pace.config.DedupConfig;

import eu.dnetlib.pace.model.Person;

public class SimpleDedupPersonMapper extends TableMapper<Text, ImmutableBytesWritable> {

	private DedupConfig dedupConf;

	private Text rowKey;

	private ImmutableBytesWritable ibw;

	@Override

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

		dedupConf = DedupConfig.load(context.getConfiguration().get(JobParams.DEDUP_CONF));

		rowKey = new Text();

		ibw = new ImmutableBytesWritable();

	}

	@Override

	protected void map(final ImmutableBytesWritable keyIn, final Result result, final Context context) throws IOException, InterruptedException {

		// System.out.println("got key: " + new String(keyIn.copyBytes()));

		if (DedupUtils.isRoot(new String(keyIn.copyBytes()))) {

			context.getCounter(dedupConf.getWf().getEntityType(), "roots skipped").increment(1);

			return;

		}

		final byte[] body = result.getValue(dedupConf.getWf().getEntityType().getBytes(), DedupUtils.BODY_B);

		if (body != null) {

			try {

				final OafDecoder decoder = OafDecoder.decode(body);

				final String hash = new Person(decoder.getEntity().getPerson().getMetadata().getFullname().getValue(), false).hash();

				// String hash = new Person(getPersonName(decoder), true).hash();

				rowKey.set(hash);

				ibw.set(body);

				context.write(rowKey, ibw);

			} catch (final Throwable e) {

				System.out.println("GOT EX " + e);

				e.printStackTrace(System.err);

				context.getCounter(dedupConf.getWf().getEntityType(), e.getClass().toString()).increment(1);

			}

		} else {

			context.getCounter(dedupConf.getWf().getEntityType(), "missing body").increment(1);

		}

	}

	// private String getPersonName(OafDecoder decoder) {

	// Metadata m = decoder.getEntity().getPerson().getMetadata();

	// String secondnames = Joiner.on(" ").join(m.getSecondnamesList());

	//

	// return isValid(m.getFullname()) ? m.getFullname() : (secondnames + ", " + m.getFirstname());

	// }

	// private boolean isValid(String fullname) {

	// return fullname != null && !fullname.isEmpty();

	// }

}