/ - Diff - D-Net - D-Net project tracking tool

modules/dnet-mapreduce-jobs/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/dedup/FindPersonCoauthorsReducer.java
1		package eu.dnetlib.data.mapreduce.hbase.dedup;
2
3		public class FindPersonCoauthorsReducer {
4
5		}

     //package eu.dnetlib.data.mapreduce.hbase.dedup;
     //
     //import java.io.IOException;
     //import java.util.List;
     //import java.util.NavigableMap;
     //import java.util.stream.Collectors;
     //
     //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.http.HttpResponse;
     //import org.apache.http.client.HttpClient;
     //import org.apache.http.client.methods.HttpPost;
     //import org.apache.http.client.methods.HttpUriRequest;
     //import org.apache.http.impl.client.DefaultHttpClient;
     //import org.apache.http.params.BasicHttpParams;
     //import org.apache.http.params.HttpParams;
     //
     //import com.google.common.collect.Lists;
     //import com.google.gson.Gson;
     //
     //import eu.dnetlib.data.mapreduce.hbase.VolatileColumnFamily;
     //import eu.dnetlib.data.proto.RelTypeProtos.RelType;
     //
     //public class FindPersonCoauthorsMapper extends TableMapper<Text, Text> {
     //
     //	private final HttpClient client = new DefaultHttpClient();
     //
     //	private final String url = "http://146.48.87.97:8888/addData";
     //
     //	@Override
     //	protected void setup(final Context context) {
     //		// url = context.getConfiguration().get("dedup.person.coauthors.service.url");
     //	}
     //
     //	@Override
     //	protected void map(final ImmutableBytesWritable rowkey, final Result row, final Context context) throws IOException, InterruptedException {
     //		final NavigableMap<byte[], byte[]> candidates = row.getFamilyMap(Bytes.toBytes(VolatileColumnFamily.dedupPerson.toString()));
     //		if ((candidates == null) || candidates.isEmpty()) return;
     //
     //		final List<String> coauthors = Lists.newArrayList();
     //		final byte[] family = Bytes.toBytes(RelType.personPublication.toString());
     //		coauthors.addAll(row.getFamilyMap(family).keySet().stream().map(Bytes::toString).collect(Collectors.toList()));
     //
     //		for (final byte[] candidate : candidates.keySet()) {
     //			emit(context, Bytes.toString(candidate), coauthors);
     //		}
     //	}
     //
     //	private void emit(final Context context, final String candidate, final List<String> coauthors) {
     //		try {
     //			final HttpUriRequest request = new HttpPost(url);
     //			final HttpParams params = new BasicHttpParams();
     //			params.setParameter("id", candidate);
     //			params.setParameter("data", (new Gson().toJson(coauthors)));
     //			request.setParams(params);
     //			final HttpResponse response = client.execute(request);
     //			context.getCounter("HTTP call", "code " + response.getStatusLine().getStatusCode()).increment(1);
     //		} catch (final Exception e) {
     //			context.getCounter("HTTP call", "Exception " + e.getClass()).increment(1);
     //		}
     //	}
     //}

     //package eu.dnetlib.data.mapreduce.util;
     //
     import java.nio.ByteBuffer;
     import eu.dnetlib.data.proto.DNGFProtos.DNGFRel;
     import eu.dnetlib.data.proto.DNGFProtos.DNGFRel.Builder;
     import eu.dnetlib.data.proto.FieldTypeProtos.Qualifier;
     import eu.dnetlib.data.proto.TypeProtos.Type;
     import eu.dnetlib.pace.config.DedupConfig;
     import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
     import org.apache.hadoop.hbase.util.Bytes;
     public class DedupUtils {
     	public static final String CF_SEPARATOR = "_";
     	public static final String ROOT = "dedup_wf";
     	public static final String BODY_S = "body";
     	public static final byte[] BODY_B = Bytes.toBytes(BODY_S);
     	public static String dedupPrefix(final String dedupRun) {
     		return "|" + ROOT + "_" + dedupRun + "::";
+    	}
     	public static String newId(final String id, final String dedupRun) {
     		if ((dedupRun == null) || (dedupRun.length() != 3)) throw new IllegalArgumentException("wrong dedupRun param");
     		return id.replaceFirst("\\|.*\\:\\:", dedupPrefix(dedupRun));
+    	}
     	public static byte[] newIdBytes(final String s, final String dedupRun) {
     		return newId(s, dedupRun).getBytes();
+    	}
     	public static byte[] newIdBytes(final ByteBuffer b, final String dedupRun) {
     		return newId(new String(b.array()), dedupRun).getBytes();
+    	}
     	public static boolean isRoot(final String s) {
     		return s.contains(ROOT);
+    	}
     	public static boolean isRoot(final ImmutableBytesWritable s) {
     		return isRoot(s.copyBytes());
+    	}
     	public static boolean isRoot(final byte[] s) {
     		return isRoot(new String(s));
+    	}
     	public static String getDedupCF_merges(final Type type) {
     		return getRelType(type) + CF_SEPARATOR + "dedup" + CF_SEPARATOR + "merges";
+    	}
     	public static String getDedupCF_merges(final String type) {
     		return getDedupCF_merges(Type.valueOf(type));
+    	}
     	public static byte[] getDedupCF_mergesBytes(final Type type) {
     		return Bytes.toBytes(getDedupCF_merges(type));
+    	}
     	public static byte[] getDedupCF_mergesBytes(final String type) {
     		return getDedupCF_mergesBytes(Type.valueOf(type));
+    	}
     	public static String getDedupCF_mergedIn(final Type type) {
     		return getRelType(type) + CF_SEPARATOR + "dedup" + CF_SEPARATOR + "isMergedIn";
+    	}
     	public static String getDedupCF_mergedIn(final String type) {
     		return getDedupCF_mergedIn(Type.valueOf(type));
+    	}
     	public static byte[] getDedupCF_mergedInBytes(final Type type) {
     		return Bytes.toBytes(getDedupCF_mergedIn(type));
+    	}
     	public static byte[] getDedupCF_mergedInBytes(final String type) {
     		return getDedupCF_mergedInBytes(Type.valueOf(type));
+    	}
     	public static String getSimilarityCF(final Type type) {
     		return getRelType(type) + CF_SEPARATOR + "dedupSimilarity" + CF_SEPARATOR + "isSimilarTo";
+    	}
     	public static String getSimilarityCF(final String type) {
     		return getSimilarityCF(Type.valueOf(type));
+    	}
     	public static byte[] getSimilarityCFBytes(final Type type) {
     		return Bytes.toBytes(getSimilarityCF(type));
+    	}
     	public static byte[] getSimilarityCFBytes(final String type) {
     		return getSimilarityCFBytes(Type.valueOf(type));
+    	}
     	public static String getRelTypeString(final Type type) {
     		return getRelType(type).toString();
+    	}
     	public static String getRelType(final Type type) {
     		switch (type) {
     		case organization:
     			return "organizationOrganization";
     		case person:
     			return "personPerson";
     		case publication:
     			return "publicationPublication";
     		case dataset:
     			return "datasetDataset";
     		default:
     			throw new IllegalArgumentException("Deduplication not supported for entity type: " + type);
+    		}
+    	}
     	public static ColumnFamily decodeCF(final byte[] b) {
     		final String[] s = new String(b).split(CF_SEPARATOR);
     		return new DedupUtils().getCF(  RelType.valueOf(s[0]), SubRelType.valueOf(s[1]));
+    	}
     	private ColumnFamily getCF(final RelType relType, final SubRelType subRelType) {
     		return new ColumnFamily(relType, subRelType);
+    	}
     	public static DNGFRel.Builder getDedup(final DedupConfig dedupConf, final String from, final String to, final String relClass) {
     		final Type type = Type.valueOf(dedupConf.getWf().getEntityType());
     		RelDescriptor descriptor = new RelDescriptor(relClass);
     		final Builder oafRel =
     				DNGFRel.newBuilder()
     						.setRelType(Qualifier.newBuilder()
     															.setClassid(descriptor.getTermCode()).setClassname(descriptor.getTermCode())
     															.setSchemeid(descriptor.getOntologyCode()).setSchemename(descriptor.getOntologyCode()))
     						.setSource(new String(from)).setSourceType(DNGFRowKeyDecoder.decode(new String(from)).getType())
     						.setTarget(new String(to)).setTargetType(DNGFRowKeyDecoder.decode(new String(to)).getType())
     						.setChild(false);
     		switch (type) {
     		case organization:
     			oafRel.setOrganizationOrganization(OrganizationOrganization.newBuilder().setDedup(
     					DedupUtils.dedup(relClass, "dnet:organization_organization_relations")));
     			break;
     		case person:
     			oafRel.setPersonPerson(PersonPerson.newBuilder().setDedup(DedupUtils.dedup(relClass, "dnet:person_person_relations")));
     			break;
     		case publication:
     			oafRel.setPublicationPublication(PublicationPublication.newBuilder().setDedup(DedupUtils.dedup(relClass, "dnet:publication_publication_relations")));
     			break;
     		case dataset:
     			oafRel.setDatasetDataset(DatasetDataset.newBuilder().setDedup(DedupUtils.dedup(relClass, "dnet:dataset_dataset_relations")));
     			break;
     		default:
     			throw new IllegalArgumentException("Deduplication not supported for entity type: " + dedupConf.getWf().getEntityType());
+    		}
     		return oafRel;
+    	}
     	private static Dedup.Builder dedup(final Dedup.RelName relClass, final String scheme) {
     		return Dedup.newBuilder().setRelMetadata(
     				RelMetadata.newBuilder().setSemantics(
     						Qualifier.newBuilder().setClassid(relClass.toString()).setClassname(relClass.toString()).setSchemeid(scheme).setSchemename(scheme)));
+    	}
     	class ColumnFamily {
     		private final RelType relType;
     		private final SubRelType subRelType;
     		public ColumnFamily(final RelType relType, final SubRelType subRelType) {
     			this.relType = relType;
     			this.subRelType = subRelType;
+    		}
     		@Override
     		public String toString() {
     			return getRelType() + CF_SEPARATOR + getSubRelType();
+    		}
     		public RelType getRelType() {
     			return relType;
+    		}
     		public SubRelType getSubRelType() {
     			return subRelType;
+    		}
+    	}
+    }

     package eu.dnetlib.data.transform;
     import java.io.IOException;
     import java.util.List;
     import java.util.concurrent.atomic.AtomicInteger;
     import eu.dnetlib.data.graph.model.DNGFDecoder;
     import eu.dnetlib.data.proto.WdsDatasetProtos.WdsDataset;
     import eu.dnetlib.data.proto.WdsDatasetProtos.WdsDataset.GeoLocation;
     import org.apache.commons.io.IOUtils;
     import org.apache.commons.lang3.StringUtils;
     import org.apache.commons.math.stat.descriptive.DescriptiveStatistics;
     import org.apache.hadoop.conf.Configuration;
     import org.apache.hadoop.hbase.client.*;
     import org.apache.hadoop.hbase.util.Bytes;
     import org.junit.*;
     import org.springframework.core.io.ClassPathResource;
     import org.springframework.core.io.Resource;
     /**
      * Created by claudio on 05/09/16.
      */
     public class HBaseReadTest {
     	private static final String TABLE_NAME = "db_wds";
     	private Resource confIn = new ClassPathResource("eu/dnetlib/data/hadoop/config/hadoop-default.dm.cnr.properties");
     	private HTable table;
     	@Before
     	public void setUp() throws IOException {
     		final Configuration conf = new Configuration();
     		for(final String line : IOUtils.readLines(confIn.getInputStream())) {
     			System.out.println("line = " + line);
     			if (!line.trim().isEmpty() && !line.startsWith("#")) {
     				final String[] split = line.split("=");
     				conf.set(split[0].trim(), split[1].trim());
+    			}
+    		}
     		table = new HTable(conf, Bytes.toBytes(TABLE_NAME));
+    	}
     	@Ignore
     	@Test
     	public void testReadGeoLocations() throws IOException {
     		final Scan scan = new Scan();
     		scan.addColumn(Bytes.toBytes("dataset"), Bytes.toBytes("body"));
     		final ResultScanner rs = table.getScanner(scan);
     		System.out.println("start iteration");
     		final DescriptiveStatistics statN = new DescriptiveStatistics();
     		final AtomicInteger invalid = new AtomicInteger(0);
     		rs.forEach(r -> {
     			final byte[] b = r.getValue(Bytes.toBytes("dataset"), Bytes.toBytes("body"));
     			final DNGFDecoder d = DNGFDecoder.decode(b, WdsDataset.geolocation);
     			final List<GeoLocation> geoList = d.getDNGF().getEntity().getDataset().getMetadata().getExtension(WdsDataset.geolocation);
     			geoList.forEach(g -> g.getBoxList().forEach(box -> {
     				if (StringUtils.isNotBlank(box)) {
     					final String[] split = box.trim().split(" ");
     					try {
     						statN.addValue(split.length);
     						Assert.assertTrue("bad number of coordinates", split.length == 4);
     						// Rect(minX=-180.0,maxX=180.0,minY=-90.0,maxY=90.0)
     						Assert.assertTrue("minX=-180", Double.parseDouble(split[1]) >= -180.0);
     						Assert.assertTrue("maxX= 180", Double.parseDouble(split[3]) <=  180.0);
     						Assert.assertTrue("minY= -90", Double.parseDouble(split[0]) >= -90.0);
     						Assert.assertTrue("maxY=  90", Double.parseDouble(split[2]) <=  90.0);
     						//maxY must be >= minY: 90.0 to -90.0
     						Assert.assertTrue("maxY must be >= minY", Double.parseDouble(split[2]) >= Double.parseDouble(split[0]));
     						//maxY must be >= minY: 90.0 to -90.0
     						Assert.assertTrue("maxX must be >= minX", Double.parseDouble(split[3]) >= Double.parseDouble(split[1]));
     					} catch (AssertionError e) {
     						invalid.set(invalid.get() + 1);
     						//System.err.println(String.format("document %s has %s coordinates: %s", d.getDNGF().getEntity().getId(), split.length, e.getMessage()));
     						//throw e;
+    					}
+    				}
     			}));
     		});
     		rs.close();
     		System.out.println(String.format("stat N: %s", statN));
     		System.out.println(String.format("invalid N: %s", invalid.get()));
+    	}
     	@After
     	public void tearDown() throws IOException {
     		table.close();
+    	}
+    }

     dnet.clustername				=	DM
     #CORE-SITE
     fs.defaultFS					=	hdfs://nmis-hadoop-cluster
     hadoop.security.authentication	=	simple
     hadoop.security.auth_to_local	=	DEFAULT
     hadoop.rpc.socket.factory.class.default	=	org.apache.hadoop.net.StandardSocketFactory
     #HBASE-SITE
     hbase.rootdir					=	hdfs://nmis-hadoop-cluster/hbase
     hbase.security.authentication	=	simple
     zookeeper.znode.rootserver		=	root-region-server
     hbase.zookeeper.quorum			=	quorum1.t.hadoop.research-infrastructures.eu,quorum2.t.hadoop.research-infrastructures.eu,quorum3.t.hadoop.research-infrastructures.eu,quorum4.t.hadoop.research-infrastructures.eu,jobtracker.t.hadoop.research-infrastructures.eu
     hbase.zookeeper.property.clientPort	=	2182
     hbase.zookeeper.client.port		=	2182
     zookeeper.znode.parent			=	/hbase
     #HDFS-SITE
     dfs.replication					=	2
     dfs.nameservices						=	nmis-hadoop-cluster
     dfs.ha.namenodes.nmis-hadoop-cluster	=	nn1,nn2
     dfs.namenode.rpc-address.nmis-hadoop-cluster.nn1=quorum1.t.hadoop.research-infrastructures.eu:8020
     dfs.namenode.http-address.nmis-hadoop-cluster.nn1=quorum1.t.hadoop.research-infrastructures.eu:50070
     dfs.namenode.rpc-address.nmis-hadoop-cluster.nn2=quorum2.t.hadoop.research-infrastructures.eu:8020
     dfs.namenode.http-address.nmis-hadoop-cluster.nn2=quorum2.t.hadoop.research-infrastructures.eu:50070
     dfs.client.failover.proxy.provider.nmis-hadoop-cluster=org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider
     #MAPRED-SITE
     mapred.job.tracker					=	nmis-hadoop-jt
     mapred.jobtrackers.nmis-hadoop-jt	=	jt1,jt2
     mapred.jobtracker.rpc-address.nmis-hadoop-jt.jt1 = jobtracker.t.hadoop.research-infrastructures.eu:8021
     mapred.jobtracker.rpc-address.nmis-hadoop-jt.jt2 = quorum4.t.hadoop.research-infrastructures.eu:8022
     mapred.client.failover.proxy.provider.nmis-hadoop-jt = org.apache.hadoop.mapred.ConfiguredFailoverProxyProvider
     mapred.mapper.new-api			=	true
     mapred.reducer.new-api			=	true
     #OOZIE SERVER
     oozie.service.loc				=	http://oozie.t.hadoop.research-infrastructures.eu:11000/oozie

     //package eu.dnetlib.data.mapreduce.hbase.dedup.experiment;
     //
     //import java.io.IOException;
     //import java.util.List;
     //
     //import eu.dnetlib.data.mapreduce.util.DNGFDecoder;
     //import eu.dnetlib.data.mapreduce.util.DedupUtils;
     //import eu.dnetlib.data.proto.FieldTypeProtos.StringField;
     //import eu.dnetlib.data.proto.PublicationProtos.Publication;
     //import org.apache.commons.lang3.StringUtils;
     //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.io.NullWritable;
     //
     ///**
     // * Created by claudio on 22/04/16.
     // */
     //public class PublicationAnalysisMapper extends TableMapper<NullWritable, NullWritable> {
     //
     //	public static final String RESULT = "result";
     //	private static final int MAX_DESCRIPTIONS = 50;
     //
     //	@Override
     //	protected void setup(final Context context) throws IOException, InterruptedException {
     //		super.setup(context);
     //	}
     //
     //	@Override
     //	protected void map(final ImmutableBytesWritable key, final Result value, final Context context) throws IOException, InterruptedException {
     //
     //		if (new String(key.copyBytes()).contains("dedup_wf")) {
     //			context.getCounter(RESULT, "roots").increment(1);
     //			return;
     //		}
     //
     //		final byte[] body = value.getValue(RESULT.getBytes(), DedupUtils.BODY_B);
     //		if (body == null) {
     //			context.getCounter(RESULT, "missing body").increment(1);
     //			return;
     //		}
     //		final DNGFDecoder decoder = DNGFDecoder.decode(body);
     //		final Publication publication = decoder.getEntity().getPublication();
     //		if (publication.getMetadata().getResulttype().getClassid().equals("dataset")) {
     //			context.getCounter(RESULT, "dataset").increment(1);
     //			return;
     //		} else {
     //			context.getCounter(RESULT, "publication").increment(1);
     //		}
     //
     //		if (publication.getMetadata().getDescriptionCount() > MAX_DESCRIPTIONS) {
     //			context.getCounter(RESULT, "abstracts > " + MAX_DESCRIPTIONS).increment(1);
     //		} else {
     //			context.getCounter(RESULT, "abstracts: " + publication.getMetadata().getDescriptionCount()).increment(1);
     //		}
     //
     //		final List<StringField> descList = publication.getMetadata().getDescriptionList();
     //
     //		boolean empty = true;
     //		for(StringField desc : descList) {
     //			empty = empty && StringUtils.isBlank(desc.getValue());
     //		}
     //
     //		context.getCounter(RESULT, "empty abstract: " + empty).increment(1);
     //	}
     //
     //	@Override
     //	protected void cleanup(final Context context) throws IOException, InterruptedException {
     //		super.cleanup(context);
     //	}
     //}
     package eu.dnetlib.data.mapreduce.hbase.dedup.experiment;
     import java.io.IOException;
     import java.util.List;
     import eu.dnetlib.data.graph.model.DNGFDecoder;
     import eu.dnetlib.data.mapreduce.util.dao.HBaseTableDAO;
     import eu.dnetlib.data.proto.FieldTypeProtos.StringField;
     import eu.dnetlib.data.proto.PublicationProtos.Publication;
     import org.apache.commons.lang3.StringUtils;
     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.io.NullWritable;
     /**
      * Created by claudio on 22/04/16.
      */
     public class PublicationAnalysisMapper extends TableMapper<NullWritable, NullWritable> {
     	public static final String RESULT = "result";
     	private static final int MAX_DESCRIPTIONS = 50;
     	@Override
     	protected void setup(final Context context) throws IOException, InterruptedException {
     		super.setup(context);
+    	}
     	@Override
     	protected void map(final ImmutableBytesWritable key, final Result value, final Context context) throws IOException, InterruptedException {
     		if (new String(key.copyBytes()).contains("dedup_wf")) {
     			context.getCounter(RESULT, "roots").increment(1);
     			return;
+    		}
     		final byte[] body = value.getValue(RESULT.getBytes(), HBaseTableDAO.cfMetadataByte());
     		if (body == null) {
     			context.getCounter(RESULT, "missing body").increment(1);
     			return;
+    		}
     		final DNGFDecoder decoder = DNGFDecoder.decode(body);
     		final Publication publication = decoder.getEntity().getPublication();
     		if (publication.getMetadata().getResulttype().getClassid().equals("dataset")) {
     			context.getCounter(RESULT, "dataset").increment(1);
     			return;
     		} else {
     			context.getCounter(RESULT, "publication").increment(1);
+    		}
     		if (publication.getMetadata().getDescriptionCount() > MAX_DESCRIPTIONS) {
     			context.getCounter(RESULT, "abstracts > " + MAX_DESCRIPTIONS).increment(1);
     		} else {
     			context.getCounter(RESULT, "abstracts: " + publication.getMetadata().getDescriptionCount()).increment(1);
+    		}
     		final List<StringField> descList = publication.getMetadata().getDescriptionList();
     		boolean empty = true;
     		for(StringField desc : descList) {
     			empty = empty && StringUtils.isBlank(desc.getValue());
+    		}
     		context.getCounter(RESULT, "empty abstract: " + empty).increment(1);
+    	}
     	@Override
     	protected void cleanup(final Context context) throws IOException, InterruptedException {
     		super.cleanup(context);
+    	}
+    }

     //package eu.dnetlib.data.mapreduce.hbase.dedup;
     //
     //import java.io.IOException;
     //import java.util.Map;
     //
     //import eu.dnetlib.data.mapreduce.JobParams;
     //import eu.dnetlib.data.mapreduce.util.DedupUtils;
     //import eu.dnetlib.data.proto.DNGFProtos.DNGF;
     //import eu.dnetlib.data.proto.DNGFProtos.DNGFRel.Builder;
     //import eu.dnetlib.data.proto.KindProtos.Kind;
     //import eu.dnetlib.data.proto.SubRelProtos.Dedup;
     //import eu.dnetlib.data.proto.TypeProtos.Type;
     //import eu.dnetlib.data.transform.xml.AbstractDNetXsltFunctions;
     //import eu.dnetlib.pace.config.DedupConfig;
     //import org.apache.commons.logging.Log;
     //import org.apache.commons.logging.LogFactory;
     //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;
     //
     //public class DedupMarkDeletedEntityMapper extends TableMapper<ImmutableBytesWritable, Put> {
     //
     //	private static final Log log = LogFactory.getLog(DedupMarkDeletedEntityMapper.class);
     //
     //	private DedupConfig dedupConf;
     //
     //	@Override
     //	protected void setup(final Context context) throws IOException, InterruptedException {
     //		dedupConf = DedupConfig.load(context.getConfiguration().get(JobParams.DEDUP_CONF));
     //		log.info("dedup findRoots mapper\nwf conf: " + dedupConf.toString());
     //	}
     //
     //	@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()));
     //
     //		final Type type = Type.valueOf(dedupConf.getWf().getEntityType());
     //		final Map<byte[], byte[]> mergedIn = value.getFamilyMap(DedupUtils.getDedupCF_mergedInBytes(type));
     //
     //		if ((mergedIn != null) && !mergedIn.isEmpty()) {
     //
     //			final byte[] row = rowkey.copyBytes();
     //
     //			// marks the original body deleted
     //			emitBody(context, row, value.getValue(Bytes.toBytes(type.toString()), DedupUtils.BODY_B));
     //
     //		} else {
     //			context.getCounter(type.toString(), "row not merged").increment(1);
     //		}
     //	}
     //
     //
     //	private void emitBody(final Context context, final byte[] row, final byte[] body) throws IOException, InterruptedException {
     //		final String type = dedupConf.getWf().getEntityType();
     //		if (body == null) {
     //			context.getCounter(type, "missing body").increment(1);
     //			System.err.println("missing body: " + new String(row));
     //			return;
     //		}
     //		final DNGF prototype = DNGF.parseFrom(body);
     //
     //		if (prototype.getDataInfo().getDeletedbyinference()) {
     //			context.getCounter(type, "bodies already deleted").increment(1);
     //		} else {
     //			final DNGF.Builder oafRoot = DNGF.newBuilder(prototype);
     //			oafRoot.getDataInfoBuilder().setDeletedbyinference(true).setInferred(true).setInferenceprovenance(dedupConf.getWf().getConfigurationId());
     //			final byte[] family = Bytes.toBytes(type);
     //			final Put put = new Put(row).add(family, DedupUtils.BODY_B, oafRoot.build().toByteArray());
     //			put.setWriteToWAL(JobParams.WRITE_TO_WAL);
     //			context.write(new ImmutableBytesWritable(row), put);
     //			context.getCounter(type, "bodies marked deleted").increment(1);
     //		}
     //	}
     //
     //	private byte[] buildRel(final byte[] from, final byte[] to, final Dedup.RelName relClass) {
     //		final Builder oafRel = DedupUtils.getDedup(dedupConf, new String(from), new String(to), relClass);
     //		final DNGF oaf =
     //				DNGF.newBuilder()
     //						.setKind(Kind.relation)
     //						.setLastupdatetimestamp(System.currentTimeMillis())
     //						.setDataInfo(
     //								AbstractDNetXsltFunctions.getDataInfo(null, "", "0.8", false, true).setInferenceprovenance(
     //										dedupConf.getWf().getConfigurationId())).setRel(oafRel)
     //				.build();
     //		return oaf.toByteArray();
     //	}
     //
     //}
     package eu.dnetlib.data.mapreduce.hbase.dedup;
     import java.io.IOException;
     import java.util.Map;
     import eu.dnetlib.data.mapreduce.JobParams;
     import eu.dnetlib.data.mapreduce.util.dao.HBaseTableDAO;
     import eu.dnetlib.data.proto.DNGFProtos.DNGF;
     import eu.dnetlib.data.proto.TypeProtos.Type;
     import eu.dnetlib.pace.config.DedupConfig;
     import org.apache.commons.logging.Log;
     import org.apache.commons.logging.LogFactory;
     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;
     public class DedupMarkDeletedEntityMapper extends TableMapper<ImmutableBytesWritable, Put> {
     	private static final Log log = LogFactory.getLog(DedupMarkDeletedEntityMapper.class);
     	private DedupConfig dedupConf;
     	@Override
     	protected void setup(final Context context) throws IOException, InterruptedException {
     		dedupConf = DedupConfig.load(context.getConfiguration().get(JobParams.DEDUP_CONF));
     		log.info("dedup findRoots mapper\nwf conf: " + dedupConf.toString());
+    	}
     	@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()));
     		final Type type = Type.valueOf(dedupConf.getWf().getEntityType());
     		final Map<byte[], byte[]> mergedIn = value.getFamilyMap(HBaseTableDAO.getDedupQualifier_mergedInBytes(type));
     		if ((mergedIn != null) && !mergedIn.isEmpty()) {
     			final byte[] row = rowkey.copyBytes();
     			// marks the original body deleted
     			emitBody(context, row, value.getValue(Bytes.toBytes(type.toString()), HBaseTableDAO.cfMetadataByte()));
     		} else {
     			context.getCounter(type.toString(), "row not merged").increment(1);
+    		}
+    	}
     	private void emitBody(final Context context, final byte[] row, final byte[] body) throws IOException, InterruptedException {
     		final String type = dedupConf.getWf().getEntityType();
     		if (body == null) {
     			context.getCounter(type, "missing body").increment(1);
     			System.err.println("missing body: " + new String(row));
     			return;
+    		}
     		final DNGF prototype = DNGF.parseFrom(body);
     		if (prototype.getDataInfo().getDeletedbyinference()) {
     			context.getCounter(type, "bodies already deleted").increment(1);
     		} else {
     			final DNGF.Builder oafRoot = DNGF.newBuilder(prototype);
     			oafRoot.getDataInfoBuilder().setDeletedbyinference(true).setInferred(true).setInferenceprovenance(dedupConf.getWf().getConfigurationId());
     			final byte[] family = Bytes.toBytes(type);
     			final Put put = new Put(row).add(family, HBaseTableDAO.cfMetadataByte(), oafRoot.build().toByteArray());
     			put.setWriteToWAL(JobParams.WRITE_TO_WAL);
     			context.write(new ImmutableBytesWritable(row), put);
     			context.getCounter(type, "bodies marked deleted").increment(1);
+    		}
+    	}
+    }

     import java.util.Map.Entry;
     import com.google.common.collect.Maps;
     import eu.dnetlib.data.graph.model.DNGFDecoder;
     import eu.dnetlib.data.mapreduce.util.dao.HBaseTableDAO;
     import eu.dnetlib.data.proto.WdsDatasetProtos.WdsDataset;
     import org.apache.hadoop.hbase.util.Bytes;
     import org.apache.hadoop.mapreduce.Mapper.Context;
-...
     	private static DNGF doMerge(final Context context, final Map<String, byte[]> map)
     			throws InvalidProtocolBufferException {
     		final byte[] value = map.get(DedupUtils.BODY_S);
     		final byte[] value = map.get(HBaseTableDAO.cfMetadata());
     		if (value == null) return null;
     		DNGF.Builder builder = DNGF.newBuilder(DNGFDecoder.decode(value, WdsDataset.geolocation).getDNGF());

     import eu.dnetlib.data.mapreduce.hbase.index.config.ContextMapper;
     import eu.dnetlib.data.mapreduce.hbase.index.config.EntityConfigTable;
     import eu.dnetlib.data.mapreduce.hbase.index.config.LinkDescriptor;
     import eu.dnetlib.data.mapreduce.hbase.index.config.RelClasses;
     import eu.dnetlib.data.proto.DNGFProtos;
     import eu.dnetlib.data.proto.FieldTypeProtos;
     import eu.dnetlib.data.proto.PublicationProtos;
-...
         protected List<DNGFDecoder> children = Lists.newLinkedList();
         protected EntityConfigTable entityConfigTable;
         protected ContextMapper contextMapper;
         protected RelClasses relClasses;
         protected Map<String, Integer> counters = Maps.newHashMap();
         protected boolean entityDefaults;
         protected boolean relDefaults;
-...
             this.relDefaults = relDefaults;
             this.childDefaults = childDefaults;
             this.ontologies = ontologies;
+        }
         public Map<String, Integer> getRelCounters() {
             return relCounters;
+        }
         public RelClasses getRelClasses() {
             return relClasses;
+        }
         public void setMainEntity(final DNGFDecoder mainEntity) {
             this.mainEntity = mainEntity;
             this.key = mainEntity.decodeEntity().getId();

     //package eu.dnetlib.data.mapreduce.hbase.dedup;
     //
     //import java.io.IOException;
     //import java.util.Map;
     //
     //import org.apache.hadoop.hbase.client.Delete;
     //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.io.Writable;
     //
     //import eu.dnetlib.data.mapreduce.JobParams;
     //import eu.dnetlib.data.mapreduce.util.DedupUtils;
     //import eu.dnetlib.data.proto.TypeProtos.Type;
     //import eu.dnetlib.pace.config.DedupConfig;
     //
     //public class DedupDeleteSimRelMapper extends TableMapper<ImmutableBytesWritable, Writable> {
     //
     //	private DedupConfig dedupConf;
     //
     //	private ImmutableBytesWritable outKey;
     //
     //	@Override
     //	protected void setup(final Context context) throws IOException, InterruptedException {
     //		dedupConf = DedupConfig.load(context.getConfiguration().get(JobParams.DEDUP_CONF));
     //		System.out.println("dedup findRoots mapper\nwf conf: " + dedupConf.toString());
     //
     //		outKey = 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()));
     //
     //		final Type type = Type.valueOf(dedupConf.getWf().getEntityType());
     //
     //		final Map<byte[], byte[]> similarRels = value.getFamilyMap(DedupUtils.getSimilarityCFBytes(type));
     //
     //		if ((similarRels != null) && !similarRels.isEmpty()) {
     //
     //			final byte[] row = rowkey.copyBytes();
     //			final Delete delete = new Delete(row);
     //
     //			for (final byte[] q : similarRels.keySet()) {
     //				delete.deleteColumns(DedupUtils.getSimilarityCFBytes(type), q);
     //			}
     //
     //			outKey.set(row);
     //			context.write(outKey, delete);
     //			context.getCounter(dedupConf.getWf().getEntityType(), "similarity deleted").increment(similarRels.size());
     //
     //		} else {
     //			context.getCounter(dedupConf.getWf().getEntityType(), "row not in similarity mesh").increment(1);
     //		}
     //	}
     //}
     package eu.dnetlib.data.mapreduce.hbase.dedup;
     import java.io.IOException;
     import java.util.Map;
     import eu.dnetlib.data.mapreduce.util.dao.HBaseTableDAO;
     import org.apache.hadoop.hbase.client.Delete;
     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.io.Writable;
     import eu.dnetlib.data.mapreduce.JobParams;
     import eu.dnetlib.data.proto.TypeProtos.Type;
     import eu.dnetlib.pace.config.DedupConfig;
     public class DedupDeleteSimRelMapper extends TableMapper<ImmutableBytesWritable, Writable> {
     	private DedupConfig dedupConf;
     	private ImmutableBytesWritable outKey;
     	@Override
     	protected void setup(final Context context) throws IOException, InterruptedException {
     		dedupConf = DedupConfig.load(context.getConfiguration().get(JobParams.DEDUP_CONF));
     		System.out.println("dedup findRoots mapper\nwf conf: " + dedupConf.toString());
     		outKey = 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()));
     		final Type type = Type.valueOf(dedupConf.getWf().getEntityType());
     		final Map<byte[], byte[]> similarRels = value.getFamilyMap(HBaseTableDAO.getSimilarityQualifierBytes(type));
     		if ((similarRels != null) && !similarRels.isEmpty()) {
     			final byte[] row = rowkey.copyBytes();
     			final Delete delete = new Delete(row);
     			for (final byte[] q : similarRels.keySet()) {
     				delete.deleteColumns(HBaseTableDAO.getSimilarityQualifierBytes(type), q);
+    			}
     			outKey.set(row);
     			context.write(outKey, delete);
     			context.getCounter(dedupConf.getWf().getEntityType(), "similarity deleted").increment(similarRels.size());
     		} else {
     			context.getCounter(dedupConf.getWf().getEntityType(), "row not in similarity mesh").increment(1);
+    		}
+    	}
+    }

     //package eu.dnetlib.data.mapreduce.hbase.dedup.cc;
     //
     //import java.io.IOException;
     //import java.nio.ByteBuffer;
     //import java.util.Set;
     //
     //import com.google.common.collect.Sets;
     //import eu.dnetlib.data.mapreduce.JobParams;
     //import eu.dnetlib.data.mapreduce.util.DedupUtils;
     //import eu.dnetlib.data.proto.DNGFProtos.DNGF;
     //import eu.dnetlib.data.proto.DNGFProtos.DNGFRel;
     //import eu.dnetlib.data.proto.KindProtos.Kind;
     //import eu.dnetlib.data.proto.SubRelProtos.Dedup;
     //import eu.dnetlib.data.proto.TypeProtos.Type;
     //import eu.dnetlib.data.transform.xml.AbstractDNetXsltFunctions;
     //import eu.dnetlib.pace.config.DedupConfig;
     //import org.apache.commons.logging.Log;
     //import org.apache.commons.logging.LogFactory;
     //import org.apache.hadoop.hbase.client.Put;
     //import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
     //import org.apache.hadoop.hbase.mapreduce.TableReducer;
     //import org.apache.hadoop.hbase.util.Bytes;
     //import org.apache.hadoop.io.Text;
     //
     ///**
     // * Created by claudio on 15/10/15.
     // */
     //public class ConnectedComponentsReducer extends TableReducer<Text, VertexWritable, ImmutableBytesWritable> {
     //
     //	private static final Log log = LogFactory.getLog(ConnectedComponentsReducer.class);
     //
     //	private DedupConfig dedupConf;
     //
     //	private byte[] cfMergedIn;
     //
     //	private byte[] cfMerges;
     //
     //	@Override
     //	protected void setup(final Context context) {
     //
     //		dedupConf = DedupConfig.load(context.getConfiguration().get(JobParams.DEDUP_CONF));
     //		log.info("dedup findRoots mapper\nwf conf: " + dedupConf.toString());
     //
     //		final Type type = Type.valueOf(dedupConf.getWf().getEntityType());
     //		cfMergedIn = DedupUtils.getDedupCF_mergedInBytes(type);
     //		cfMerges = DedupUtils.getDedupCF_mergesBytes(type);
     //	}
     //
     //	@Override
     //	protected void reduce(Text key, Iterable<VertexWritable> values, Context context) throws IOException, InterruptedException {
     //
     //		final Set<String> set = Sets.newHashSet();
     //
     //		for(VertexWritable v : values) {
     //			for(Text t : v.getEdges()) {
     //				set.add(t.toString());
     //			}
     //		}
     //
     //		final byte[] root = DedupUtils.newIdBytes(ByteBuffer.wrap(Bytes.toBytes(key.toString())), dedupConf.getWf().getDedupRun());
     //
     //		for(String q : set) {
     //			final byte[] qb = Bytes.toBytes(q);
     //			emitDedupRel(context, cfMergedIn, qb, root, buildRel(qb, root, Dedup.RelName.isMergedIn));
     //			emitDedupRel(context, cfMerges, root, qb, buildRel(root, qb, Dedup.RelName.merges));
     //
     //			context.getCounter(dedupConf.getWf().getEntityType(), "dedupRel (x2)").increment(1);
     //		}
     //
     //	}
     //
     //	private void emitDedupRel(final Context context, final byte[] cf, final byte[] from, final byte[] to, final byte[] value) throws IOException,
     //			InterruptedException {
     //		final Put put = new Put(from).add(cf, to, value);
     //		put.setWriteToWAL(JobParams.WRITE_TO_WAL);
     //		context.write(new ImmutableBytesWritable(from), put);
     //	}
     //
     //	private byte[] buildRel(final byte[] from, final byte[] to, final Dedup.RelName relClass) {
     //		final DNGFRel.Builder oafRel = DedupUtils.getDedup(dedupConf, new String(from), new String(to), relClass);
     //		final DNGF oaf =
     //				DNGF.newBuilder()
     //						.setKind(Kind.relation)
     //						.setLastupdatetimestamp(System.currentTimeMillis())
     //						.setDataInfo(
     //								AbstractDNetXsltFunctions.getDataInfo(null, "", "0.8", false, true).setInferenceprovenance(
     //										dedupConf.getWf().getConfigurationId())).setRel(oafRel)
     //						.build();
     //		return oaf.toByteArray();
     //	}
     //
     //}
     package eu.dnetlib.data.mapreduce.hbase.dedup.cc;
     import java.io.IOException;
     import java.nio.ByteBuffer;
     import java.util.Set;
     import com.google.common.collect.Sets;
     import eu.dnetlib.data.mapreduce.JobParams;
     import eu.dnetlib.data.mapreduce.util.dao.HBaseTableDAO;
     import eu.dnetlib.data.proto.DNGFProtos.DNGF;
     import eu.dnetlib.data.proto.DNGFProtos.DNGFRel;
     import eu.dnetlib.data.proto.KindProtos.Kind;
     import eu.dnetlib.data.proto.TypeProtos.Type;
     import eu.dnetlib.data.transform.xml.AbstractDNetXsltFunctions;
     import eu.dnetlib.pace.config.DedupConfig;
     import org.apache.commons.logging.Log;
     import org.apache.commons.logging.LogFactory;
     import org.apache.hadoop.hbase.client.Put;
     import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
     import org.apache.hadoop.hbase.mapreduce.TableReducer;
     import org.apache.hadoop.hbase.util.Bytes;
     import org.apache.hadoop.io.Text;
     /**
      * Created by claudio on 15/10/15.
      */
     public class ConnectedComponentsReducer extends TableReducer<Text, VertexWritable, ImmutableBytesWritable> {
     	private static final Log log = LogFactory.getLog(ConnectedComponentsReducer.class);
     	private DedupConfig dedupConf;
     	private byte[] cfMergedIn;
     	private byte[] cfMerges;
     	@Override
     	protected void setup(final Context context) {
     		dedupConf = DedupConfig.load(context.getConfiguration().get(JobParams.DEDUP_CONF));
     		log.info("dedup findRoots mapper\nwf conf: " + dedupConf.toString());
     		final Type type = Type.valueOf(dedupConf.getWf().getEntityType());
     		cfMergedIn = HBaseTableDAO.getDedupQualifier_mergedInBytes(type);
     		cfMerges = HBaseTableDAO.getDedupQualifier_mergesBytes(type);
+    	}
     	@Override
     	protected void reduce(Text key, Iterable<VertexWritable> values, Context context) throws IOException, InterruptedException {
     		final Set<String> set = Sets.newHashSet();
     		for(VertexWritable v : values) {
     			for(Text t : v.getEdges()) {
     				set.add(t.toString());
+    			}
+    		}
     		final byte[] root = HBaseTableDAO.newIdBytes(ByteBuffer.wrap(Bytes.toBytes(key.toString())), dedupConf.getWf().getDedupRun());
     		for(String q : set) {
     			final byte[] qb = Bytes.toBytes(q);
     			emitDedupRel(context, cfMergedIn, qb, root, buildRel(qb, root, "isMergedIn"));
     			emitDedupRel(context, cfMerges, root, qb, buildRel(root, qb, "merges"));
     			context.getCounter(dedupConf.getWf().getEntityType(), "dedupRel (x2)").increment(1);
+    		}
+    	}
     	private void emitDedupRel(final Context context, final byte[] cf, final byte[] from, final byte[] to, final byte[] value) throws IOException,
     			InterruptedException {
     		final Put put = new Put(from).add(cf, to, value);
     		put.setWriteToWAL(JobParams.WRITE_TO_WAL);
     		context.write(new ImmutableBytesWritable(from), put);
+    	}
     	private byte[] buildRel(final byte[] from, final byte[] to, final String relClass) {
     		final DNGFRel.Builder oafRel = HBaseTableDAO.getDedup(new String(from), new String(to), relClass);
     		final DNGF oaf =
     				DNGF.newBuilder()
     						.setKind(Kind.relation)
     						.setLastupdatetimestamp(System.currentTimeMillis())
     						.setDataInfo(
     								AbstractDNetXsltFunctions.getDataInfo(null, "", "0.8", false, true).setInferenceprovenance(
     										dedupConf.getWf().getConfigurationId())).setRel(oafRel)
     						.build();
     		return oaf.toByteArray();
+    	}
+    }

     //package eu.dnetlib.data.mapreduce.hbase.dedup;
     //
     //import java.io.IOException;
     //import java.util.Collection;
     //import java.util.List;
     //import java.util.Map;
     //
     //import com.google.common.collect.Maps;
     //import eu.dnetlib.data.mapreduce.JobParams;
     //import eu.dnetlib.data.mapreduce.util.DedupUtils;
     //import eu.dnetlib.data.mapreduce.util.DNGFDecoder;
     //import eu.dnetlib.data.proto.DNGFProtos.DNGFEntity;
     //import eu.dnetlib.data.proto.TypeProtos.Type;
     //import eu.dnetlib.pace.clustering.BlacklistAwareClusteringCombiner;
     //import eu.dnetlib.pace.config.DedupConfig;
     //import eu.dnetlib.pace.model.MapDocument;
     //import eu.dnetlib.pace.model.ProtoDocumentBuilder;
     //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;
     //
     //public class DedupMapper extends TableMapper<Text, ImmutableBytesWritable> {
     //
     //	private static final Log log = LogFactory.getLog(DedupMapper.class);
     //
     //	private DedupConfig dedupConf;
     //
     //	private Map<String, List<String>> blackListMap = Maps.newHashMap();
     //
     //	private Text outKey;
     //
     //	private ImmutableBytesWritable ibw;
     //
     //	@Override
     //	protected void setup(final Context context) throws IOException, InterruptedException {
     //
     //		final String dedupConfJson = context.getConfiguration().get(JobParams.DEDUP_CONF);
     //
     //		log.info("pace conf strings");
     //		log.info("pace conf: " + dedupConfJson);
     //
     //		dedupConf = DedupConfig.load(dedupConfJson);
     //
     //		blackListMap = dedupConf.getPace().getBlacklists();
     //
     //		outKey = new Text();
     //		ibw = new ImmutableBytesWritable();
     //
     //		log.info("pace conf");
     //		log.info("entity type: " + dedupConf.getWf().getEntityType());
     //		log.info("clustering: " + dedupConf.getPace().getClustering());
     //		log.info("conditions: " + dedupConf.getPace().getConditions());
     //		log.info("fields: " + dedupConf.getPace().getModel());
     //		log.info("blacklists: " + blackListMap);
     //		log.info("wf conf: " + dedupConf.toString());
     //	}
     //
     //	@Override
     //	protected void map(final ImmutableBytesWritable keyIn, final Result result, final Context context) throws IOException, InterruptedException {
     //		// log.info("got key: " + new String(keyIn.copyBytes()));
     //
     //		final byte[] body = result.getValue(dedupConf.getWf().getEntityType().getBytes(), DedupUtils.BODY_B);
     //
     //		if (body != null) {
     //
     //			final DNGFDecoder decoder = DNGFDecoder.decode(body);
     //			if (decoder.getDNGF().getDataInfo().getDeletedbyinference()) {
     //				context.getCounter(dedupConf.getWf().getEntityType(), "deleted by inference").increment(1);
     //				return;
     //			}
     //
     //			final DNGFEntity entity = decoder.getEntity();
     //
     //			context.getCounter(entity.getType().toString(), "decoded").increment(1);
     //
     //			if (entity.getType().equals(Type.valueOf(dedupConf.getWf().getEntityType()))) {
     //
     //				// GeneratedMessage metadata = DNGFEntityDecoder.decode(entity).getEntity();
     //				final MapDocument doc = ProtoDocumentBuilder.newInstance(Bytes.toString(keyIn.copyBytes()), entity, dedupConf.getPace().getModel());
     //				emitNGrams(context, doc, BlacklistAwareClusteringCombiner.filterAndCombine(doc, dedupConf, blackListMap));
     //			}
     //		} else {
     //			context.getCounter(dedupConf.getWf().getEntityType(), "missing body").increment(1);
     //		}
     //	}
     //
     //	private void emitNGrams(final Context context, final MapDocument doc, final Collection<String> ngrams) throws IOException, InterruptedException {
     //		for (final String ngram : ngrams) {
     //			outKey.set(ngram);
     //			ibw.set(doc.toByteArray());
     //			context.write(outKey, ibw);
     //		}
     //	}
     //
     //}
     package eu.dnetlib.data.mapreduce.hbase.dedup;
     import java.io.IOException;
     import java.util.Collection;
     import java.util.List;
     import java.util.Map;
     import com.google.common.collect.Maps;
     import eu.dnetlib.data.graph.model.DNGFDecoder;
     import eu.dnetlib.data.mapreduce.JobParams;
     import eu.dnetlib.data.mapreduce.util.dao.HBaseTableDAO;
     import eu.dnetlib.data.proto.DNGFProtos.DNGFEntity;
     import eu.dnetlib.data.proto.TypeProtos.Type;
     import eu.dnetlib.pace.clustering.BlacklistAwareClusteringCombiner;
     import eu.dnetlib.pace.config.DedupConfig;
     import eu.dnetlib.pace.model.MapDocument;
     import eu.dnetlib.pace.model.ProtoDocumentBuilder;
     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;
     public class DedupMapper extends TableMapper<Text, ImmutableBytesWritable> {
     	private static final Log log = LogFactory.getLog(DedupMapper.class);
     	private DedupConfig dedupConf;
     	private Map<String, List<String>> blackListMap = Maps.newHashMap();
     	private Text outKey;
     	private ImmutableBytesWritable ibw;
     	@Override
     	protected void setup(final Context context) throws IOException, InterruptedException {
     		final String dedupConfJson = context.getConfiguration().get(JobParams.DEDUP_CONF);
     		log.info("pace conf strings");
     		log.info("pace conf: " + dedupConfJson);
     		dedupConf = DedupConfig.load(dedupConfJson);
     		blackListMap = dedupConf.getPace().getBlacklists();
     		outKey = new Text();
     		ibw = new ImmutableBytesWritable();
     		log.info("pace conf");
     		log.info("entity type: " + dedupConf.getWf().getEntityType());
     		log.info("clustering: " + dedupConf.getPace().getClustering());
     		log.info("conditions: " + dedupConf.getPace().getConditions());
     		log.info("fields: " + dedupConf.getPace().getModel());
     		log.info("blacklists: " + blackListMap);
     		log.info("wf conf: " + dedupConf.toString());
+    	}
     	@Override
     	protected void map(final ImmutableBytesWritable keyIn, final Result result, final Context context) throws IOException, InterruptedException {
     		// log.info("got key: " + new String(keyIn.copyBytes()));
     		final byte[] body = result.getValue(dedupConf.getWf().getEntityType().getBytes(), HBaseTableDAO.cfMetadataByte());
     		if (body != null) {
     			final DNGFDecoder decoder = DNGFDecoder.decode(body);
     			if (decoder.getDNGF().getDataInfo().getDeletedbyinference()) {
     				context.getCounter(dedupConf.getWf().getEntityType(), "deleted by inference").increment(1);
     				return;
+    			}
     			final DNGFEntity entity = decoder.getEntity();
     			context.getCounter(entity.getType().toString(), "decoded").increment(1);
     			if (entity.getType().equals(Type.valueOf(dedupConf.getWf().getEntityType()))) {
     				// GeneratedMessage metadata = DNGFEntityDecoder.decode(entity).getEntity();
     				final MapDocument doc = ProtoDocumentBuilder.newInstance(Bytes.toString(keyIn.copyBytes()), entity, dedupConf.getPace().getModel());
     				emitNGrams(context, doc, BlacklistAwareClusteringCombiner.filterAndCombine(doc, dedupConf, blackListMap));
+    			}
     		} else {
     			context.getCounter(dedupConf.getWf().getEntityType(), "missing body").increment(1);
+    		}
+    	}
     	private void emitNGrams(final Context context, final MapDocument doc, final Collection<String> ngrams) throws IOException, InterruptedException {
     		for (final String ngram : ngrams) {
     			outKey.set(ngram);
     			ibw.set(doc.toByteArray());
     			context.write(outKey, ibw);
+    		}
+    	}
+    }

     		</dependency>
     		<dependency>
     			<groupId>org.apache.commons</groupId>
     			<artifactId>commons-lang3</artifactId>
     			<version>${commons.lang.version}</version>
     		</dependency>
     		<dependency>
     			<groupId>com.typesafe</groupId>
     			<artifactId>config</artifactId>
     			<version>1.3.0</version>
-...
     			<artifactId>mongo-java-driver</artifactId>
     			<version>${mongodb.driver.version}</version>
     		</dependency>
                     <dependency>
             <dependency>
     			<groupId>org.elasticsearch</groupId>
     			<artifactId>elasticsearch-hadoop-mr</artifactId>
     			<version>2.0.2</version>
-...
     			<version>1.8.5</version>
     			<scope>test</scope>
     		</dependency>
     		<dependency>
     			<groupId>org.apache.zookeeper</groupId>
     			<artifactId>zookeeper</artifactId>
     			<version>3.4.5-cdh4.3.0</version>
     			<scope>test</scope>
     		</dependency>
     		<dependency>
     			<groupId>org.json</groupId>
     			<artifactId>json</artifactId>
     			<version>20160212</version>

     //package eu.dnetlib.data.mapreduce.hbase.dedup;
     //
     //import java.io.IOException;
     //
     //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.io.Text;
     //
     //import eu.dnetlib.data.mapreduce.JobParams;
     //import eu.dnetlib.data.mapreduce.util.DedupUtils;
     //import eu.dnetlib.data.mapreduce.util.DNGFDecoder;
     //import eu.dnetlib.pace.config.DedupConfig;
     //import eu.dnetlib.pace.model.Person;
     //
     //public class SimpleDedupPersonMapper extends TableMapper<Text, ImmutableBytesWritable> {
     //
     //	private DedupConfig dedupConf;
     //
     //	private Text rowKey;
     //
     //	private ImmutableBytesWritable ibw;
     //
     //	@Override
     //	protected void setup(final Context context) throws IOException, InterruptedException {
     //		dedupConf = DedupConfig.load(context.getConfiguration().get(JobParams.DEDUP_CONF));
     //		rowKey = new Text();
     //		ibw = new ImmutableBytesWritable();
     //	}
     //
     //	@Override
     //	protected void map(final ImmutableBytesWritable keyIn, final Result result, final Context context) throws IOException, InterruptedException {
     //		// System.out.println("got key: " + new String(keyIn.copyBytes()));
     //
     //		if (DedupUtils.isRoot(new String(keyIn.copyBytes()))) {
     //			context.getCounter(dedupConf.getWf().getEntityType(), "roots skipped").increment(1);
     //			return;
     //		}
     //		final byte[] body = result.getValue(dedupConf.getWf().getEntityType().getBytes(), DedupUtils.BODY_B);
     //
     //		if (body != null) {
     //			try {
     //				final DNGFDecoder decoder = DNGFDecoder.decode(body);
     //
     //				final String hash = new Person(decoder.getEntity().getPerson().getMetadata().getFullname().getValue(), false).hash();
     //				// String hash = new Person(getPersonName(decoder), true).hash();
     //
     //				rowKey.set(hash);
     //				ibw.set(body);
     //				context.write(rowKey, ibw);
     //
     //			} catch (final Throwable e) {
     //				System.out.println("GOT EX " + e);
     //				e.printStackTrace(System.err);
     //				context.getCounter(dedupConf.getWf().getEntityType(), e.getClass().toString()).increment(1);
     //			}
     //		} else {
     //			context.getCounter(dedupConf.getWf().getEntityType(), "missing body").increment(1);
     //		}
     //	}
     //
     //	// private String getPersonName(DNGFDecoder decoder) {
     //	// Metadata m = decoder.getEntity().getPerson().getMetadata();
     //	// String secondnames = Joiner.on(" ").join(m.getSecondnamesList());
     //	//
     //	// return isValid(m.getFullname()) ? m.getFullname() : (secondnames + ", " + m.getFirstname());
     //	// }
     //
     //	// private boolean isValid(String fullname) {
     //	// return fullname != null && !fullname.isEmpty();
     //	// }
     //
     //}
     package eu.dnetlib.data.mapreduce.hbase.dedup;
     import java.io.IOException;
     import eu.dnetlib.data.graph.model.DNGFDecoder;
     import eu.dnetlib.data.mapreduce.util.dao.HBaseTableDAO;
     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.io.Text;
     import eu.dnetlib.data.mapreduce.JobParams;
     import eu.dnetlib.pace.config.DedupConfig;
     import eu.dnetlib.pace.model.Person;
     public class SimpleDedupPersonMapper extends TableMapper<Text, ImmutableBytesWritable> {
     	private DedupConfig dedupConf;
     	private Text rowKey;
     	private ImmutableBytesWritable ibw;
     	@Override
     	protected void setup(final Context context) throws IOException, InterruptedException {
     		dedupConf = DedupConfig.load(context.getConfiguration().get(JobParams.DEDUP_CONF));
     		rowKey = new Text();
     		ibw = new ImmutableBytesWritable();
+    	}
     	@Override
     	protected void map(final ImmutableBytesWritable keyIn, final Result result, final Context context) throws IOException, InterruptedException {
     		// System.out.println("got key: " + new String(keyIn.copyBytes()));
     		if (HBaseTableDAO.isRoot(new String(keyIn.copyBytes()))) {
     			context.getCounter(dedupConf.getWf().getEntityType(), "roots skipped").increment(1);
     			return;
+    		}
     		final byte[] body = result.getValue(dedupConf.getWf().getEntityType().getBytes(), HBaseTableDAO.cfMetadataByte());
     		if (body != null) {
     			try {
     				final DNGFDecoder decoder = DNGFDecoder.decode(body);
     				final String hash = new Person(decoder.getEntity().getPerson().getMetadata().getFullname().getValue(), false).hash();
     				// String hash = new Person(getPersonName(decoder), true).hash();
     				rowKey.set(hash);
     				ibw.set(body);
     				context.write(rowKey, ibw);
     			} catch (final Throwable e) {
     				System.out.println("GOT EX " + e);
     				e.printStackTrace(System.err);
     				context.getCounter(dedupConf.getWf().getEntityType(), e.getClass().toString()).increment(1);
+    			}
     		} else {
     			context.getCounter(dedupConf.getWf().getEntityType(), "missing body").increment(1);
+    		}
+    	}

Project

General

Profile

D-Net

Revision 45419

Added by Claudio Atzori over 7 years ago