D-Net

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

import java.io.IOException;

import java.util.Map;

import java.util.Map.Entry;

import org.apache.commons.collections.MapUtils;

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 org.apache.hadoop.mapreduce.Counter;

import com.google.protobuf.InvalidProtocolBufferException;

import eu.dnetlib.data.mapreduce.JobParams;

import eu.dnetlib.data.mapreduce.hbase.index.config.EntityConfigTable;

import eu.dnetlib.data.mapreduce.hbase.index.config.IndexConfig;

import eu.dnetlib.data.mapreduce.hbase.index.config.LinkDescriptor;

import eu.dnetlib.data.mapreduce.hbase.index.config.RelClasses;

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

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

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

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.OafProtos.OafRel;

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

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

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

public class PrepareFeedMapper extends TableMapper<Text, ImmutableBytesWritable> {

	private EntityConfigTable entityConfigTable;

	private RelClasses relClasses;

	private Text outKey;

	private ImmutableBytesWritable ibw;

	@Override

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

		final String json = context.getConfiguration().get(JobParams.INDEX_ENTITY_LINKS);

		System.out.println(JobParams.INDEX_ENTITY_LINKS + ":\n" + json);

		entityConfigTable = IndexConfig.load(json).getConfigMap();

		final String contextMap = context.getConfiguration().get("contextmap");

		System.out.println("contextmap:\n" + contextMap);

		final String relClassJson = context.getConfiguration().get("relClasses");

		System.out.println("relClassesJson:\n" + relClassJson);

		relClasses = RelClasses.fromJSon(relClassJson);

		System.out.println("relClasses:\n" + relClasses);

		outKey = new Text();

		ibw = new ImmutableBytesWritable();

	}

	@Override

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

		final OafRowKeyDecoder keyDecoder = OafRowKeyDecoder.decode(keyIn.copyBytes());

		final Type type = keyDecoder.getType();

		final Oaf oaf = mergeUpdates(value, context, type, keyDecoder);

		if (isValid(oaf)) {

			if (deletedByInference(oaf) && DedupUtils.isRoot(keyIn)) {

				incrementCounter(context, "deleted by inference (root)", type.toString(), 1);

				return;

			}

			if (!deletedByInference(oaf) || entityConfigTable.includeDuplicates(type)) {

				emit(new String(keyIn.copyBytes()), context, oaf);

				incrementCounter(context, Kind.entity.toString(), type.toString(), 1);

				for (final LinkDescriptor ld : entityConfigTable.getDescriptors(type)) {

					final Map<byte[], byte[]> columnMap = value.getFamilyMap(Bytes.toBytes(ld.getRelDescriptor().getIt()));

					if (!MapUtils.isEmpty(columnMap)) {

						emitRelationship(oaf.getEntity(), context, columnMap, ld);

						incrementCounter(context, type.toString(), ld.getRelDescriptor().getIt(), columnMap.size());

					}

				}

			} else {

				incrementCounter(context, "deleted by inference", type.toString(), 1);

			}

		} else {

			incrementCounter(context, "missing body", type.toString(), 1);

		}

	}

	private Oaf mergeUpdates(final Result value, final Context context, final Type type, final OafRowKeyDecoder keyDecoder)

			throws InvalidProtocolBufferException {

		try {

			return UpdateMerger.mergeBodyUpdates(context, value.getFamilyMap(Bytes.toBytes(type.toString())));

		} catch (final InvalidProtocolBufferException e) {

			System.err.println(String.format("Unable to parse proto (Type: %s) in row: %s", type.toString(), keyDecoder.getKey()));

			throw e;

		}

	}

	private void emitRelationship(final OafEntity cachedTarget, final Context context, final Map<byte[], byte[]> columnMap, final LinkDescriptor ld)

			throws IOException, InterruptedException {

		final Oaf.Builder oafBuilder = Oaf.newBuilder().setKind(Kind.relation);

		// iterates the column map

		for (final Entry<byte[], byte[]> e : columnMap.entrySet()) {

			final Oaf oaf = decodeProto(context, e.getValue());

			if (!isValid(oaf)) {

				incrementCounter(context, "invalid oaf rel", ld.getRelDescriptor().getIt(), 1);

			} else if (!deletedByInference(oaf)) {

				final OafRel.Builder relBuilder = OafRel.newBuilder(oaf.getRel());

				if (ld.isSymmetric()) {

					final RelDescriptor rd = ld.getRelDescriptor();

					relBuilder.setCachedTarget(cachedTarget).setRelType(rd.getRelType()).setSubRelType(rd.getSubRelType());

				}

				if (ld.getRelDescriptor().getIt().contains(SubRelType.dedup.toString()) && isDedupSelf(relBuilder)) {

					incrementCounter(context, "avoid to emit dedup self", ld.getRelDescriptor().getIt(), 1);

					continue;

				}

				final OafRel oafRel = relBuilder.setChild(ld.isChild()).build();

				// String rowKey = patchTargetId(target, OafRelDecoder.decode(oafRel).getRelTargetId());

				emit(ld.isSymmetric() ? oafRel.getTarget() : oafRel.getSource(), context, merge(oafBuilder, oaf).setRel(oafRel).build());

			} else {

				incrementCounter(context, "deleted by inference", ld.getRelDescriptor().getIt(), 1);

			}

		}

	}

	// private String patchTargetId(final Type target, final String id) {

	// return id.replaceFirst("^.*\\|", target.getNumber() + "|");

	// }

	private Oaf.Builder merge(final Oaf.Builder builder, final Oaf prototype) {

		return builder.setDataInfo(prototype.getDataInfo()).setTimestamp(prototype.getTimestamp());

	}

	private boolean isDedupSelf(final OafRelOrBuilder rel) {

		return rel.getSource().contains(rel.getTarget());

	}

	private boolean isValid(final Oaf oaf) {

		return (oaf != null) && oaf.isInitialized();

	}

	private boolean deletedByInference(final Oaf oaf) {

		return oaf.getDataInfo().getDeletedbyinference();

	}

	private Oaf decodeProto(final Context context, final byte[] body) {

		try {

			return Oaf.parseFrom(body);

		} catch (final InvalidProtocolBufferException e) {

			e.printStackTrace(System.err);

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

		}

		return null;

	}

	// private byte[] prefix(final Type name, final byte[] bytes) {

	// return concat(concat(name.toString().getBytes(), PrepareFeedJob.bSEPARATOR), bytes);

	// }

	private void emit(final String key, final Context context, final Oaf oaf) throws IOException, InterruptedException {

		// Text keyOut = new Text(Hashing.murmur3_128().hashString(key).toString());

		outKey.set(key);

		ibw.set(oaf.toByteArray());

		context.write(outKey, ibw);

	}

	// protected byte[] concat(final byte[] a, final byte[] b) {

	// byte[] c = new byte[a.length + b.length];

	// System.arraycopy(a, 0, c, 0, a.length);

	// System.arraycopy(b, 0, c, a.length, b.length);

	//

	// return c;

	// }

	private void incrementCounter(final Context context, final String k, final String t, final int n) {

		getCounter(context, k, t).increment(n);

	}

	private Counter getCounter(final Context context, final String k, final String t) {

		return context.getCounter(k, t);

	}

}