D-Net

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

import java.io.IOException;

import java.util.Set;

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.hbase.util.Bytes;

import org.apache.hadoop.io.Text;

import com.google.common.base.Function;

import com.google.common.collect.Iterables;

import com.google.common.collect.Sets;

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

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

import eu.dnetlib.data.proto.TypeProtos.Type;

import eu.dnetlib.pace.model.PersonComparatorUtils;

public class FindDedupCandidatePersonsMapper extends TableMapper<Text, Text> {

	private static final byte[] PERSON_CF = Type.person.toString().getBytes();

	private static final byte[] PERSONRESULT_CF = "personResult".getBytes();

	@Override

	protected void setup(final Context context) {

	}

	@Override

	protected void map(final ImmutableBytesWritable rowkey, final Result row, final Context context) throws IOException, InterruptedException {

		String id = Bytes.toString(rowkey.get());

		String fullname = extractFullname(row);

		Set<String> resultIds = extractResultIds(row);

		if (fullname != null) {

			Text text = (new DedupPersonBean(id, fullname, resultIds)).toText();

			for (String k : PersonComparatorUtils.getNgramsForPerson(fullname)) {

				context.write(new Text(k), text);

			}

		}

	}

	private Set<String> extractResultIds(final Result row) {

		return Sets.newHashSet(Iterables.transform(row.getFamilyMap(PERSONRESULT_CF).keySet(), new Function<byte[], String>() {

			@Override

			public String apply(final byte[] b) {

				return Bytes.toString(b);

			}

		}));

	}

	private String extractFullname(final Result row) {

		byte[] body = row.getValue(PERSON_CF, DedupUtils.BODY_B);

		if (body == null) return null;

		return OafDecoder.decode(body).getEntity().getPerson().getMetadata().getFullname().getValue();

	}

}