D-Net

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

import java.io.IOException;

import java.nio.ByteBuffer;

import java.util.Collections;

import java.util.HashSet;

import java.util.Map;

import java.util.Set;

import com.google.common.base.Function;

import com.google.common.collect.Sets;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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.JobParams;

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.config.DedupConfig;

public class DedupBuildRootsMapper extends TableMapper<Text, ImmutableBytesWritable> {

	private static final Log log = LogFactory.getLog(DedupBuildRootsMapper.class);

	private DedupConfig dedupConf;

	private ImmutableBytesWritable ibw;

	private Set<String> entityNames;

	@Override

	protected void setup(final Context context) {

		dedupConf = DedupConfig.load(context.getConfiguration().get(JobParams.DEDUP_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()));

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

		if (checkDataset(value)) return;

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

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

			final Set<String> ids = getStrings(mergedIn);

			if (ids.size() > 1) {

				context.getCounter(dedupConf.getWf().getEntityType(), "mergedIn > 1").increment(1);

			}

			final Text rootId = new Text(Collections.min(ids));

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

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

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

			for (final String family : dedupConf.getWf().getRootBuilder()) {

				// 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.getWf().getEntityType(), "not in duplicate group").increment(1);

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

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

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

				if (body != null) {

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

					emit(context, new Text(rowkey.copyBytes()), body);

				}

			}

		}

	}

	private HashSet<String> getStrings(final Map<byte[], byte[]> mergedIn) {

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

			@Override

			public String apply(final byte[] input) {

				return new String(input);

			}

		}));

	}

	private boolean checkDataset(final Result value) {

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

		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.getWf().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.getWf().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;

		}

	}

}