D-Net

mvn assembly:assembly -DskipTests=true && mvn install:install-file -Dfile=target/dnet-mapreduce-jobs-0.0.6.4-SNAPSHOT-jar-with-dependencies.jar -DgroupId=eu.dnetlib -DartifactId=dnet-mapreduce-jobs-assembly -Dversion=0.0.6.4-SNAPSHOT -Dpackaging=jar

		final String id = "id";

		final OafDecoder decoder = OafTest.embed(OafTest.getResult(id), Kind.entity);

		builder.setMainEntity(decoder);

		// System.out.println("decoded " + JsonFormat.printToString(OafDecoder.decode(Base64.decodeBase64(base64String)).getOaf()));

		final String xml = builder.build().replaceAll("\\n", " ").replaceAll("\\s+", " ").replaceAll("> <", "><");

		System.out.println("xml size: " + xml.length());

		System.out.println(xml.replaceAll("\\n", " ").replaceAll("\\s+", " ").replaceAll("> <", "><"));

    <dc:format>dc.format</dc:format>

import java.util.ArrayList;

import java.util.List;

import org.apache.commons.collections.MapUtils;

import com.google.common.base.Function;

import com.google.common.collect.Iterables;

import com.google.common.collect.Lists;

import eu.dnetlib.data.mapreduce.JobParams;

import eu.dnetlib.pace.model.RootMapDocument;

		paceConf = DynConf.load(context.getConfiguration().get(JobParams.PACE_CONF));

		dedupConf = DedupConfigLoader.load(context.getConfiguration().get(JobParams.WF_CONF));

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

			final OafDecoder decoder = OafDecoder.decode(body);

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

			if (decoder.getEntity().getType().equals(Type.valueOf(dedupConf.getEntityType()))) {

				if (decoder.getOaf().getDataInfo().getDeletedbyinference()) {

					context.getCounter(dedupConf.getEntityType(), "deleted by inference").increment(1);

					return;

				}

				if (decoder.getOaf().getDataInfo().getInferred()) {

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

				}

				if (isDataset(decoder)) return;

				final MapDocument doc = ProtoDocumentBuilder.newInstance(Bytes.toString(keyIn.copyBytes()), decoder.getEntity(), paceConf.fields());

				final List<String> merges = getMergedIds(result);

				emitNGrams(context, new RootMapDocument(doc, merges), BlacklistAwareClusteringCombiner.filterAndCombine(doc, paceConf, blackListMap));

	private List<String> getMergedIds(final Result result) {

		final Map<byte[], byte[]> mergesMap = result.getFamilyMap(DedupUtils.getDedupCF_mergesBytes(dedupConf.getEntityType()));

		return MapUtils.isEmpty(mergesMap) ? new ArrayList<String>() : Lists.newArrayList(Iterables.transform(mergesMap.keySet(),

				new Function<byte[], String>() {

					@Override

					public String apply(final byte[] id) {

						return Bytes.toString(id);

					}

				}));

	}

	private boolean isDataset(final OafDecoder decoder) {

		return decoder.getEntity().getType().equals(Type.result)

				&& decoder.getEntity().getResult().getMetadata().getResulttype().getClassid().equals("dataset");

	}

	private void emitNGrams(final Context context, final RootMapDocument doc, final Collection<String> ngrams) throws IOException, InterruptedException {

		if (doc.hasMerges()) {

			context.getCounter(dedupConf.getEntityType(), "merged").increment(doc.getMerges().size());

		}

		for (final String ngram : ngrams) {

import eu.dnetlib.data.mapreduce.JobParams;

		dedupConf = DedupConfigLoader.load(context.getConfiguration().get(JobParams.WF_CONF));

		// if (DedupUtils.isRoot(keyIn)) {

		// context.getCounter(COUNTER_GROUP, "skipped roots").increment(1);

		// return;

		// }

		for (final KeyValue n : kvList) {

			for (final KeyValue j : kvList) {

				final byte[] nq = n.getQualifier();

				final byte[] jq = j.getQualifier();

					final Put put = new Put(nq).add(DedupUtils.getSimilarityCFBytes(Type.valueOf(dedupConf.getEntityType())), jq, Bytes.toBytes(""));

					put.setWriteToWAL(JobParams.WRITE_TO_WAL);

import eu.dnetlib.data.mapreduce.JobParams;

import eu.dnetlib.pace.model.RootMapDocument;

import eu.dnetlib.pace.model.RootMapDocumentSerializer;

		paceConf = DynConf.load(context.getConfiguration().get(JobParams.PACE_CONF));

		dedupConf = DedupConfigLoader.load(context.getConfiguration().get(JobParams.WF_CONF));

		final Queue<RootMapDocument> q = prepare(context, key, values);

	private Queue<RootMapDocument> prepare(final Context context, final Text key, final Iterable<ImmutableBytesWritable> values) {

		final Queue<RootMapDocument> queue = new PriorityQueue<RootMapDocument>(100, new MapDocumentComparator(dedupConf.getOrderField()));

		for (final ImmutableBytesWritable i : values) {

			final RootMapDocument doc = RootMapDocumentSerializer.decode(i.copyBytes());

			final String id = doc.getIdentifier();

				// System.out.println("breaking out after limit (" + dedupConf.getQueueMaxSize() + ") for ngram '" + key);

	private Queue<RootMapDocument> simplifyQueue(final Queue<RootMapDocument> queue, final String ngram, final Context context) {

		final Queue<RootMapDocument> q = new LinkedList<RootMapDocument>();

		final List<RootMapDocument> tempResults = Lists.newArrayList();

			final RootMapDocument result = queue.remove();

				final String field = NGramUtils.cleanupForOrdering(result.values(dedupConf.getOrderField()).stringValue());

	private void populateSimplifiedQueue(final Queue<RootMapDocument> q,

			final List<RootMapDocument> tempResults,

					.increment(tempResults.size());

	private void process(final Queue<RootMapDocument> queue, final Context context) throws IOException, InterruptedException {

			final RootMapDocument pivot = queue.remove();

			final String fieldPivot = (fieldsPivot == null) || fieldsPivot.isEmpty() ? null : fieldsPivot.stringValue();

				for (final RootMapDocument curr : queue) {

					final String fieldCurr = (fieldsCurr == null) || fieldsCurr.isEmpty() ? null : fieldsCurr.stringValue();

					if (!idCurr.equals(idPivot) && (fieldCurr != null)) {

						final double d = algo.between(pivot, curr, paceConf);

							handleRoot(context, pivot);

							handleRoot(context, curr);

	private void handleRoot(final Context context, final RootMapDocument doc) throws IOException, InterruptedException {

		if (DedupUtils.isRoot(doc.getIdentifier())) {

			context.getCounter(dedupConf.getEntityType(), "root " + SubRelType.dedupSimilarity.toString() + " (x2)").increment(1);

			for (final String mergedId : doc.getMerges()) {

				writeSimilarity(context, doc.getIdentifier(), mergedId);

			}

		}

	}

		final byte[] rowKey = Bytes.toBytes(idPivot);

		final byte[] target = Bytes.toBytes(id);

		final Put put = new Put(from).add(DedupUtils.getSimilarityCFBytes(Type.valueOf(dedupConf.getEntityType())), to, Bytes.toBytes(""));

		put.setWriteToWAL(JobParams.WRITE_TO_WAL);

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

import java.io.IOException;

import java.nio.charset.Charset;

import java.util.List;

import java.util.Map;

import java.util.UUID;

import org.apache.commons.collections.MapUtils;

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 com.google.common.collect.Lists;

import com.googlecode.protobuf.format.JsonFormat;

import eu.dnetlib.data.mapreduce.JobParams;

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

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

import eu.dnetlib.pace.util.DedupConfig;

import eu.dnetlib.pace.util.DedupConfigLoader;

public class DedupRootsToCsvMapper extends TableMapper<ImmutableBytesWritable, ImmutableBytesWritable> {

	/**

	 * logger.

	 */

	private static final Log log = LogFactory.getLog(DedupRootsToCsvMapper.class); // NOPMD by marko on 11/24/08 5:02 PM

	private DedupConfig dedupConf;

	private ImmutableBytesWritable key;

	private ImmutableBytesWritable value;

	@Override

	protected void setup(final Context context) {

		dedupConf = DedupConfigLoader.load(context.getConfiguration().get(JobParams.WF_CONF));

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

		key = new ImmutableBytesWritable();

		value = new ImmutableBytesWritable();

	}

	@Override

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

		// if (!DedupUtils.isRoot(rowkey)) {

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

		// return;

		// }

		final Map<byte[], byte[]> entityCf = result.getFamilyMap(Bytes.toBytes(dedupConf.getEntityType()));

		if (MapUtils.isEmpty(entityCf) && (entityCf.get(DedupUtils.BODY_B) == null)) {

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

			return;

		}

		final byte[] body = entityCf.get(DedupUtils.BODY_B);

		final String joaf = JsonFormat.printToString(OafDecoder.decode(body).getOaf());

		final List<String> mergedIds = Lists.newArrayList();

		final Map<byte[], byte[]> mergedIn = result.getFamilyMap(DedupUtils.getDedupCF_mergesBytes(dedupConf.getEntityType()));

		if (MapUtils.isNotEmpty(mergedIn)) {

			for (final byte[] q : mergedIn.keySet()) {

				final String id = new String(q, Charset.forName("UTF-8"));

				// log.info("rowkey: '" + new String(rowkey.copyBytes(), Charset.forName("UTF-8")) + "' mergedId: '" + id + "'\n\n\n");

				mergedIds.add(id);

			}

		}

		if (mergedIds.isEmpty()) {

			context.getCounter(dedupConf.getEntityType(), "root with no merged ids").increment(1);

			// return;

		}

		final RootEntity re = new RootEntity(joaf, mergedIds);

		key.set(Bytes.toBytes(UUID.randomUUID().toString()));

		value.set(Bytes.toBytes(re.toString()));

		context.write(key, value);

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

	}

}

import eu.dnetlib.data.mapreduce.JobParams;

		dedupConf = DedupConfigLoader.load(context.getConfiguration().get(JobParams.WF_CONF));

			// if (isDeletedByInference(value)) {

			// context.getCounter(dedupConf.getEntityType(), "deleted by inference").increment(1);

			// return;

			// }

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

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

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

					}

	// private boolean isDeletedByInference(final Result value) {

	// final byte[] body = value.getFamilyMap(Bytes.toBytes(dedupConf.getEntityType())).get(DedupUtils.BODY_B);

	//

	// if (body == null) return false;

	//

	// return OafDecoder.decode(body).getOaf().getDataInfo().getDeletedbyinference();

	// }

			final Oaf oaf = Oaf.parseFrom(o);

		} catch (final InvalidProtocolBufferException e) {

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

import java.util.List;

import com.google.gson.Gson;

public class RootEntity {

	private String joaf;

	private List<String> mergedIds;

	public RootEntity() {}

	public RootEntity(final String joaf, final List<String> mergedIds) {

		super();

		this.setJoaf(joaf);

		this.setMergedIds(mergedIds);

	}

	public static RootEntity decode(final String s) {

		return new Gson().fromJson(s, RootEntity.class);

	}

	@Override

	public String toString() {

		return new Gson().toJson(this);

	}

	public String getJoaf() {

		return joaf;

	}

	public void setJoaf(final String joaf) {

		this.joaf = joaf;

	}

	public List<String> getMergedIds() {

		return mergedIds;

	}

	public void setMergedIds(final List<String> mergedIds) {

		this.mergedIds = mergedIds;

	}

}

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

import java.io.IOException;

import java.nio.charset.Charset;

import java.util.List;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;

import com.googlecode.protobuf.format.JsonFormat;

import com.googlecode.protobuf.format.JsonFormat.ParseException;

import eu.dnetlib.data.mapreduce.JobParams;

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

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

import eu.dnetlib.pace.util.DedupConfig;

import eu.dnetlib.pace.util.DedupConfigLoader;

public class DedupRootsToCsvReducer extends Reducer<ImmutableBytesWritable, ImmutableBytesWritable, Text, Text> {

	/**

	 * logger.

	 */

	private static final Log log = LogFactory.getLog(DedupRootsToCsvReducer.class); // NOPMD by marko on 11/24/08 5:02 PM

	private static final String COUNTER_GROUP = "csv";

	enum Tables {

		Groups, NativeGroups, NativeEntities

	}

	private DedupConfig dedupConf;

	private String DELIM;

	private String WRAP;

	private Text tKey;

	private Text tValue;

	private MultipleOutputs<Text, Text> mos;

	@Override

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

		super.setup(context);

		tKey = new Text();

		tValue = new Text();

		mos = new MultipleOutputs<Text, Text>(context);

		dedupConf = DedupConfigLoader.load(context.getConfiguration().get(JobParams.WF_CONF));

		log.info("wf conf: " + dedupConf.toString());

		DELIM = context.getConfiguration().get("mapred.textoutputformat.separator", "!");

		WRAP = context.getConfiguration().get("mapred.textoutputformat.wrapper", "#");

		log.info("unsing field DELIMITER: '" + DELIM + "'");

	}

	@Override

	protected void reduce(final ImmutableBytesWritable key, final Iterable<ImmutableBytesWritable> values, final Context context) throws IOException,

			InterruptedException {

		for (final ImmutableBytesWritable ibw : values) {

			final RootEntity rootEntity = RootEntity.decode(new String(ibw.copyBytes(), Charset.forName("UTF-8")));

			emitRoot(context, rootEntity.getJoaf());

			emitRels(context, key, rootEntity.getMergedIds());

			return;

		}

	}

	private void emitRoot(final Context context, final String joaf) throws IOException, InterruptedException {

		final Oaf.Builder builder = Oaf.newBuilder();

		try {

			JsonFormat.merge(joaf, builder);

		} catch (final ParseException e) {

			context.getCounter("roots csv", e.getClass().getSimpleName()).increment(1);

			return;

		}

		final OafDecoder d = OafDecoder.decode(builder.build());

		tKey.set((WRAP + d.getEntityId() + WRAP).getBytes(Charset.forName("UTF-8")));

		// tValue.set(value.getBytes(Charset.forName("UTF-8")));

		mos.write(Tables.NativeEntities.toString(), tKey, tValue, Tables.NativeEntities.toString());

		context.getCounter(COUNTER_GROUP, "native_entities").increment(1);

	}

	private void emitRels(final Context context, final ImmutableBytesWritable key, final List<String> mergedIds) throws IOException, InterruptedException {

		final StringBuilder sb = new StringBuilder();

		final String groupId = new String(key.copyBytes());

		// native_groups groups native_entities

		tKey.set((WRAP + groupId + WRAP).getBytes(Charset.forName("UTF-8")));

		tValue.set((WRAP + dedupConf.getConfigurationId() + WRAP).getBytes(Charset.forName("UTF-8")));

		mos.write(Tables.Groups.toString(), tKey, tValue, Tables.Groups.toString());

		context.getCounter(COUNTER_GROUP, "groups").increment(mergedIds.size());

		for (final String id : mergedIds) {

			sb.append(WRAP).append(id).append(WRAP).append(DELIM);

			tValue.set(sb.toString().getBytes(Charset.forName("UTF-8")));

			mos.write(Tables.NativeGroups.toString(), tKey, tValue, Tables.NativeGroups.toString());

		}

		context.getCounter(COUNTER_GROUP, "native_groups").increment(mergedIds.size());

	}

	@Override

	public void cleanup(final Context context) throws IOException, InterruptedException {

		mos.close();

	}

}

		dedupConf = DedupConfigLoader.load(context.getConfiguration().get(JobParams.WF_CONF));

			// System.err.println("aborting DedupBuildRootsReducer, found non-root key: " + key);

		put.setWriteToWAL(JobParams.WRITE_TO_WAL);

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

import java.io.IOException;

import java.util.Collection;

import java.util.List;

import java.util.Set;

import org.apache.commons.csv.CSVFormat;

import org.apache.commons.csv.CSVPrinter;

import org.apache.commons.lang.CharUtils;

import com.google.common.base.Function;

import com.google.common.base.Joiner;

import com.google.common.collect.Iterables;

import com.google.common.collect.Lists;

import com.google.common.collect.Sets;

import com.googlecode.protobuf.format.JsonFormat;

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

import eu.dnetlib.data.proto.FieldTypeProtos.DataInfo;

import eu.dnetlib.data.proto.FieldTypeProtos.KeyValue;

import eu.dnetlib.data.proto.FieldTypeProtos.StringField;

import eu.dnetlib.data.proto.FieldTypeProtos.StructuredProperty;

import eu.dnetlib.data.proto.OrganizationProtos.Organization;

import eu.dnetlib.data.proto.ResultProtos.Result;

public class CSVSerializer {

	private CSVFormat format;

	public CSVSerializer(final String delimiter, final String wrapper) {

		format = CSVFormat.newFormat(CharUtils.toChar(delimiter)).withQuote(CharUtils.toChar(wrapper));

	}

	public String getNativeEntity(final OafDecoder d) throws IOException {

		final StringBuilder sb = new StringBuilder();

		final CSVPrinter csv = new CSVPrinter(sb, format);

		csv.print(d.getEntityId());

		csv.print(j(u(kv(d.getEntity().getCollectedfromList()))));

		csv.print(type(d));

		csv.print(body(d));

		csv.print(ft(d));

		csv.flush();

		csv.close();

		return sb.toString();

	}

	private String type(final OafDecoder d) {

		final DataInfo dataInfo = d.getOaf().getDataInfo();

		return dataInfo.getDeletedbyinference() && dataInfo.getInferenceprovenance().contains("dedup") ? "r" : "n";

	}

	private String body(final OafDecoder d) {

		return JsonFormat.printToString(d.getMetadata());

	}

	private String ft(final OafDecoder d) {

		final StringBuilder sb = new StringBuilder();

		switch (d.getEntity().getType()) {

		case result:

			final Result.Metadata mr = d.getEntity().getResult().getMetadata();

			sb.append(sp(mr.getTitleList())).append(" ").append(sf(mr.getDescriptionList()));

			break;

		case organization:

			final Organization.Metadata om = d.getEntity().getOrganization().getMetadata();

			sb.append(om.getLegalname().getValue() + " " + om.getLegalshortname().getValue());

			break;

		case person:

		case project:

		case datasource:

		default:

			throw new IllegalArgumentException("Unhandled fulltext extraction for type: " + d.getEntity().getType());

		}

		return sb.toString();

	}

	private Set<String> u(final Collection<String> kv) {

		return Sets.newLinkedHashSet(kv);

	}

	private String j(final Collection<String> l) {

		return Joiner.on(" ").skipNulls().join(l);

	}

	private String sf(final List<StringField> list) {

		final StringBuilder sb = new StringBuilder();

		for (final StringField sp : list) {

			sb.append(sp.getValue()).append(" ");

		}

		return sb.toString().trim();

	}

	private String sp(final List<StructuredProperty> list) {

		final StringBuilder sb = new StringBuilder();

		for (final StructuredProperty sp : list) {

			sb.append(sp.getValue()).append(" ");

		}

		return sb.toString().trim();

	}

	private List<String> kv(final List<KeyValue> list) {

		return Lists.newLinkedList(Iterables.transform(list, new Function<KeyValue, String>() {

			@Override

			public String apply(final KeyValue kv) {

				return kv.getValue().trim();

			}

		}));

	}

}

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

import java.io.IOException;

import java.util.Map;

import org.apache.commons.collections.MapUtils;

import org.apache.commons.lang.StringUtils;

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

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 eu.dnetlib.data.mapreduce.JobParams;

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

public class OfflineHbaseLoadMapper extends TableMapper<ImmutableBytesWritable, Put> {

	private String entityType;

	@Override

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

		super.setup(context);

		entityType = context.getConfiguration().get("entityType");

		if (StringUtils.isBlank(entityType)) throw new IllegalArgumentException("missing entityType parameter");

	}

	@Override

	protected void map(final ImmutableBytesWritable key, final Result value, final Context context)

			throws IOException, InterruptedException {

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

		if (MapUtils.isEmpty(entityMap) || !entityMap.containsKey(DedupUtils.BODY_B)) {

			context.getCounter(entityType, "missing body").increment(1);

		}

		final byte[] body = entityMap.get(DedupUtils.BODY_B);

		final Put put = new Put(key.copyBytes());

		put.setWriteToWAL(JobParams.WRITE_TO_WAL);

		put.add(Bytes.toBytes(entityType), DedupUtils.BODY_B, body);

		context.write(key, put);

		context.getCounter(entityType, "loaded").increment(1);

	}

	@Override

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

		super.cleanup(context);

	}

}

import eu.dnetlib.data.mapreduce.JobParams;

	public static final String COUNTER_GROUP = "dedup.patch.roots";

		dedupConf = DedupConfigLoader.load(context.getConfiguration().get(JobParams.WF_CONF));

		final Type type = Type.valueOf(dedupConf.getEntityType());

		final Map<byte[], byte[]> similarRels = value.getFamilyMap(DedupUtils.getSimilarityCFBytes(type));

			if (DedupUtils.isRoot(rowkey)) {

				context.getCounter(COUNTER_GROUP, "roots").increment(1);

			}

			final ByteBuffer min = findMin(ByteBuffer.wrap(rowkey.get()), similarRels.keySet());

			final byte[] row = rowkey.copyBytes();

			final byte[] root = DedupUtils.newIdBytes(min, dedupConf.getDedupRun());

		for (final byte[] q : keys) {

			final ByteBuffer iq = ByteBuffer.wrap(q);

			// System.err.println("missing body: " + new String(row));

			final Oaf.Builder oafRoot = Oaf.newBuilder(prototype);

			final byte[] family = Bytes.toBytes(dedupConf.getEntityType());

			final Put put = new Put(row).add(family, DedupUtils.BODY_B, oafRoot.build().toByteArray());

			put.setWriteToWAL(JobParams.WRITE_TO_WAL);

		final Builder oafRel = DedupUtils.getDedup(dedupConf, new String(from), new String(to), relClass);

		final Oaf oaf =

						.setDataInfo(AbstractDNetOafXsltFunctions.getDataInfo(null, "", "0.8", false, true).setInferenceprovenance("dedup")).setRel(oafRel)

				.build();

			InterruptedException {

		final Put put = new Put(from).add(cf, to, value);

		put.setWriteToWAL(JobParams.WRITE_TO_WAL);

import org.apache.commons.collections.MapUtils;

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

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

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

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

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

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

				return;

			}

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

					if (!MapUtils.isEmpty(columnMap)) {

					}

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

			throws InvalidProtocolBufferException {

		} catch (final InvalidProtocolBufferException e) {

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

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

					final RelDescriptor rd = ld.getRelDescriptor();

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

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

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

	// }

		} catch (final InvalidProtocolBufferException e) {

	public static String getDedupCF_merges(final String type) {

		return getDedupCF_merges(Type.valueOf(type));

	}

	public static byte[] getDedupCF_mergesBytes(final String type) {

		return getDedupCF_mergesBytes(Type.valueOf(type));

	}

	public static String getDedupCF_mergedIn(final String type) {

		return getDedupCF_mergedIn(Type.valueOf(type));

	}

	public static byte[] getDedupCF_mergedInBytes(final String type) {

		return getDedupCF_mergedInBytes(Type.valueOf(type));

	}

	public static String getSimilarityCF(final String type) {

		return getSimilarityCF(Type.valueOf(type));

	}

	public static byte[] getSimilarityCFBytes(final String type) {

		return getSimilarityCFBytes(Type.valueOf(type));

	}

		final String[] s = new String(b).split(CF_SEPARATOR);

		final Type type = Type.valueOf(dedupConf.getEntityType());

		final RelType relType = DedupUtils.getRelType(type);

		final Builder oafRel =

					throws TransformerConfigurationException, TransformerFactoryConfigurationError {

		final OafEntityDecoder entity = mainEntity.decodeEntity();

		final List<String> rels = listRelations();

		final DataInfo dataInfo = decoder.getOaf().getDataInfo();

		final StringBuilder sb = new StringBuilder();

		if ((decoder.getEntity() instanceof Result) && !expandingRel) {

		if ((decoder.getEntity() instanceof Person) && !expandingRel) {

		if ((decoder.getEntity() instanceof Project) && !expandingRel) {

		final LicenseComparator lc = new LicenseComparator();

		for (final Instance instance : ((Result) mainEntity.decodeEntity().getEntity()).getInstanceList()) {

		for (final OafDecoder decoder : this.relations) {

				final Type targetType = relDecoder.getTargetType(mainEntity.getEntity().getType());

				final Set<String> relFilter = entityConfigTable.getFilter(targetType, relDecoder.getRelDescriptor());

				final RelDescriptor relDescriptor = relDecoder.getRelDescriptor();

				if ((cachedTarget != null) && cachedTarget.isInitialized()) {

				final RelMetadata relMetadata = relDecoder.getRelMetadata();

					semanticclass = semanticscheme = "";

				final LinkDescriptor ld = entityConfigTable.getDescriptor(relDecoder.getTargetType(mainEntity.getEntity().getType()), relDescriptor);

				final String relId = (ld != null) && !ld.isSymmetric() ? relDecoder.getRelTargetId() : relDecoder.getRelSourceId();