D-Net

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

import java.io.IOException;

import java.util.Map;

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.collect.Iterables;

import com.google.protobuf.InvalidProtocolBufferException;

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

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

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

import eu.dnetlib.data.proto.KindProtos.Kind;

import eu.dnetlib.data.proto.OafProtos.Oaf;

import eu.dnetlib.data.proto.OafProtos.OafEntity;

import eu.dnetlib.data.proto.RelTypeProtos.RelType;

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

import eu.dnetlib.data.transform.OafUtils;

import eu.dnetlib.pace.util.DedupConfig;

import eu.dnetlib.pace.util.DedupConfigLoader;

public class DedupBuildRootsMapper extends TableMapper<Text, ImmutableBytesWritable> {

	private DedupConfig dedupConf;

	private ImmutableBytesWritable ibw;

	private Set<String> entityNames;

	@Override

	protected void setup(final Context context) {

		dedupConf = DedupConfigLoader.load(context.getConfiguration().get("dedup.wf.conf"));

		System.out.println("dedup buildRoots mapper\nwf conf: " + dedupConf.toString());

		entityNames = OafUtils.entities();

		ibw = new ImmutableBytesWritable();

	}

	@Override

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

		// System.out.println("Find root mapping: " + new String(rowkey.copyBytes()));

		if (!DedupUtils.isRoot(rowkey)) {

			// TODO: remove this hack - here because we don't want to dedup datasets

			if (checkDataset(value)) return;

			final Map<byte[], byte[]> dedupRels = value.getFamilyMap(DedupUtils.getDedupCF_mergedInBytes(Type.valueOf(dedupConf.getEntityType())));

			if ((dedupRels != null) && !dedupRels.isEmpty()) {

				final Text rootId = findRoot(dedupRels);

				// byte[] rootIdBytes = rootId.copyBytes();

				// byte[] rowkeyBytes = rowkey.copyBytes();

				context.getCounter(dedupConf.getEntityType(), "merged").increment(1);

				for (final String family : dedupConf.getRootBuilderFamilies()) {

					// if (checkHack(rowkeyBytes, rootIdBytes, family)) {

					// context.getCounter("hack", "personResult skipped").increment(1);

					// } else {

					final Map<byte[], byte[]> map = value.getFamilyMap(Bytes.toBytes(family));

					if ((map != null) && !map.isEmpty()) {

						// if we're dealing with the entity CF

						if (entityNames.contains(family)) {

							final Oaf body = UpdateMerger.mergeBodyUpdates(context, map);

							emit(context, rootId, body.toByteArray());

						} else {

							for (final byte[] o : map.values()) {

								if (!isRelMarkedDeleted(context, o)) {

									emit(context, rootId, o);

								} else {

									context.getCounter(family, "rel marked deleted").increment(1);

								}

							}

						}

					} // else {

					// System.err.println("empty family: " + family + "\nkey: " + sKey);

					// context.getCounter("DedupBuildRootsMapper", "empty family '" + family + "'").increment(1);

					// }

					// }

				}

			} else {

				context.getCounter(dedupConf.getEntityType(), "not in duplicate group").increment(1);

			}

		} else {

			context.getCounter(dedupConf.getEntityType(), "root").increment(1);

		}

	}

	private boolean checkDataset(final Result value) {

		final Map<byte[], byte[]> bodyMap = value.getFamilyMap(dedupConf.getEntityNameBytes());

		if ((bodyMap == null) || bodyMap.isEmpty()) return true;

		final byte[] bodyB = bodyMap.get(DedupUtils.BODY_B);

		if (bodyB == null) return true;

		final OafEntity entity = OafDecoder.decode(bodyB).getEntity();

		if (entity.getType().equals(Type.result) && entity.getResult().getMetadata().getResulttype().getClassid().equals("dataset")) return true;

		return false;

	}

	private void emit(final Context context, final Text rootId, final byte[] value) throws IOException, InterruptedException {

		ibw.set(value);

		context.write(rootId, ibw);

	}

	private boolean checkHack(final byte[] rowkey, final byte[] rootId, final String family) {

		return dedupConf.getEntityType().equals(Type.result.toString()) && 	// we're deduplicating the results

				family.equals(RelType.personResult.toString()) && 			// we're dealing with a personResult relation

				!rootId.equals(DedupUtils.newIdBytes(new String(rowkey), dedupConf.getDedupRun()));

	}

	private boolean isRelMarkedDeleted(final Context context, final byte[] o) {

		try {

			final Oaf oaf = Oaf.parseFrom(o);

			return oaf.getKind().equals(Kind.relation) && oaf.getDataInfo().getDeletedbyinference();

		} catch (final InvalidProtocolBufferException e) {

			context.getCounter("error", e.getClass().getName()).increment(1);

			return true;

		}

	}

	private Text findRoot(final Map<byte[], byte[]> dedupRels) {

		return new Text(Iterables.getOnlyElement(dedupRels.keySet()));

	}

}