D-Net

{"type_source": "SVN", "goal": "package -U -T 4C source:jar", "url": "http://svn-public.driver.research-infrastructures.eu/driver/dnet45/modules/dnet-mapreduce-jobs/trunk/", "deploy_repository": "dnet45-snapshots", "version": "4", "mail": "sandro.labruzzo@isti.cnr.it,michele.artini@isti.cnr.it, claudio.atzori@isti.cnr.it, alessia.bardi@isti.cnr.it", "deploy_repository_url": "http://maven.research-infrastructures.eu/nexus/content/repositories/dnet45-snapshots", "name": "dnet-mapreduce-jobs"}

net.sf.saxon.TransformerFactoryImpl

^(Corpus Oral Dialectal \(COD\)\.).*$

^(Kiri Karl Morgensternile).*$

^(\[Eksliibris Aleksandr).*\]$

^(Kiri A\. de Vignolles).*$

^(2 kirja Karl Morgensternile).*$

^(Pirita kloostri idaosa arheoloogilised).*$

^(Kiri tundmatule).*$

^(Kiri Jenaer Allgemeine Literaturzeitung toimetusele).*$

^(Eksliibris Nikolai Birukovile).*$

^(Eksliibris Nikolai Issakovile).*$

^(\[Eksliibris Aleksandr).*$

^(WHP Cruise Summary Information of section).*$

^(Measurement of the top quark\-pair production cross section with ATLAS in pp collisions at).*$

^(Measurement of the spin\-dependent structure function).*

<oaf:$name$>

	$metadata:{$it$}$

  <rels>

    $rels:{$it$}$

  </rels>

  <children>

	$children:{$it$}$

  </children>

</oaf:$name$>

$inference:{$it$}$

<?xml version="1.0"?>

<record>

  <result xmlns:dri="http://www.driver-repository.eu/namespace/dri" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">

    <header>

      <dri:objIdentifier>$id$</dri:objIdentifier>

      <dri:dateOfCollection>$dateofcollection$</dri:dateOfCollection>

      <dri:dateOfTransformation>$dateoftransformation$</dri:dateOfTransformation>

      <counters>

	  $counters:{$it$}$

	  </counters>

    </header>

    <metadata>

      <oaf:entity xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 

		    xmlns:oaf="http://namespace.openaire.eu/oaf" 

		    xsi:schemaLocation="http://namespace.openaire.eu/oaf $schemaLocation$">

		$it$

      </oaf:entity>

    </metadata>

  </result>

</record>

<result>

    <dri:objIdentifier>$objIdentifier$</dri:objIdentifier>

	$metadata:{$it$}$

</result>

<rel inferred="$inferred$" trust="$trust$" inferenceprovenance="$inferenceprovenance$" provenanceaction="$provenanceaction$">

  <to class="$class$" scheme="$scheme$" type="$type$">$objIdentifier$</to>

  $metadata:{$it$}$

</rel>

<$name$$if(hasId)$ objidentifier="$id$"$else$$endif$>

	$metadata:{$it$}$

</$name$>

<instance id="$instanceId$">

    $metadata:{$it$}$

	$webresources:{$it$}$

</instance>

<webresource>

  <url>$identifier$</url>

</webresource>

### Root Level ###

log4j.rootLogger=WARN, CONSOLE

### Configuration for the CONSOLE appender ###

log4j.appender.CONSOLE=org.apache.log4j.ConsoleAppender

log4j.appender.CONSOLE.layout=org.apache.log4j.PatternLayout

log4j.appender.CONSOLE.layout.ConversionPattern=[%-5p] %d %c - %m%n

### Application Level ###

log4j.logger.eu.dnetlib=INFO

log4j.logger.eu.dnetlib.enabling.is.sn=INFO

log4j.logger.org.apache.cxf.interceptor=FATAL

log4j.logger.org.apache.cxf.ws.addressing.ContextUtils=FATAL

log4j.logger.eu.dnetlib.enabling.tools.AbstractBaseService=INFO

log4j.logger.eu.dnetlib.enabling.inspector=DEBUG

log4j.logger.eu.dnetlib.xml.database.LoggingTrigger=WARN

log4j.logger.eu.dnetlib.enabling.tools.registration.ServiceRegistrator=INFO

log4j.logger.eu.dnetlib.enabling.inspector=FATAL

log4j.logger.eu.dnetlib.enabling.inspector.SubscriptionController=DEBUG

log4j.logger.eu.dnetlib.springutils.stringtemplate.StringTemplateViewResolver=FATAL

log4j.logger.eu.dnetlib.enabling.is.sn.SynchronousNotificationSenderImpl=WARN

log4j.logger.eu.dnetlib.enabling.tools.LocalServiceResolverImpl=WARN

log4j.logger.eu.dnetlib.enabling.is.sn.NotificationInvokerImpl=WARN

log4j.logger.eu.dnetlib.data.collective=INFO

log4j.logger.eu.dnetlib.data.hadoop.utils.ScanFactory=DEBUG

log4j.logger.org.apache.xerces.parsers.SAXParser=OFF

log4j.logger.eu.dnetlib.conf.PropertyFetcher=WARN

#log4j.logger.eu.dnetlib.data.transform.XsltRowTransformerFactory=DEBUG

log4j.logger.eu.dnetlib.enabling.is.sn.ISSNServiceImpl=OFF

log4j.logger.eu.dnetlib.enabling.datasources.DatasourceManagerClients=FATAL

log4j.logger.eu.dnetlib.data.mdstore.modular.mongodb.utils.MetadataCheckJob=DEBUG

log4j.logger.eu.dnetlib.enabling.is.sn.ISSNServiceCore=WARN

log4j.logger.eu.dnetlib.xml.database.exist.ExistDatabase=WARN

log4j.logger.eu.dnetlib.enabling.is.store.AbstractContentInitializer=FATAL

log4j.logger.org.apache.hadoop.hbase.mapreduce.TableInputFormatBase=FATAL

log4j.logger.eu.dnetlib.data.mdstore.modular.plugin.CreatorExtractor=DEBUG

### Spring ###

log4j.logger.org.springframework=ERROR

log4j.logger.org.apache=DEBUG

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

import com.google.common.base.Splitter;

import com.google.common.collect.Lists;

import eu.dnetlib.data.bulktag.CommunityConfiguration;

import eu.dnetlib.data.bulktag.CommunityConfigurationFactory;

import eu.dnetlib.data.proto.FieldTypeProtos;

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

import eu.dnetlib.data.proto.ResultProtos;

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 org.apache.hadoop.io.Writable;

import java.io.IOException;

import java.util.List;

import java.util.Map;

public class BulkTaggingMapper extends TableMapper<ImmutableBytesWritable, Writable> {

	private CommunityConfiguration cc;

	private ResultTagger tagger;

	private boolean enabled;

	@Override

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

		super.setup(context);

		final String conf = context.getConfiguration().get("tagging.conf");

		enabled = context.getConfiguration().getBoolean("tagging.enabled",false);

		if (StringUtils.isBlank(conf)) {

			throw new IllegalArgumentException("missing bulk tagging configuration");

		}

		System.out.println("conf = " + conf);

		cc = CommunityConfigurationFactory.fromJson(conf);

		tagger = new ResultTagger();

		tagger.setTrust(context.getConfiguration().get("bulktagging.trust", "0.85"));

	}

	@Override

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

		final Map<byte[], byte[]> resultMap = value.getFamilyMap(Bytes.toBytes("result"));

		final byte[] body = resultMap.get(Bytes.toBytes("body"));

		if (body != null) {

			context.getCounter("Bulk Tagging", "not null body ").increment(1);

			final Oaf oaf = tagger.enrichContext(Oaf.parseFrom(body), cc, context);

			if (oaf == null) {

				//context.getCounter("In mapper", " null oaf ").increment(1);

				return;

			}

			long tagged = oaf.getEntity().getResult().getMetadata().getContextList().stream()

					.flatMap(c -> c.getDataInfoList().stream())

					.map(FieldTypeProtos.DataInfo::getInferenceprovenance)

					.filter(infProv -> "bulktagging".equals(infProv))

					.count();

			context.getCounter("Bulk Tagging", " bulktagged ").increment(tagged);

			final Put put = new Put(key.copyBytes()).add(Bytes.toBytes("result"), Bytes.toBytes("body"), oaf.toByteArray());

			if(tagged > 0){

				if (enabled)

					context.write(key, put);

				context.getCounter("Bulk Tagging", " write op ").increment(1);

			}

		}

		else{

			context.getCounter("Bulk Tagging", " null body ").increment(1);

		}

	}

}

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

import com.google.common.base.Functions;

import com.google.common.collect.Maps;

import com.google.common.collect.Sets;

import eu.dnetlib.data.bulktag.CommunityConfiguration;

import eu.dnetlib.data.bulktag.Pair;

import eu.dnetlib.data.proto.FieldTypeProtos;

import eu.dnetlib.data.proto.OafProtos;

import eu.dnetlib.data.proto.ResultProtos;

import org.apache.commons.lang3.StringUtils;

import org.apache.hadoop.mapreduce.Mapper;

import java.util.*;

import java.util.stream.Collectors;

import java.util.stream.Stream;

/**

 * Created by miriam on 02/08/2018.

 */

public class ResultTagger {

    private final static String DATA_INFO_TYPE = "bulktagging";

    private final static String SCHEMA_NAME = "dnet:provenanceActions";

    private final static String CLASS_ID = "bulktagging::community";

    private final static String SCHEMA_ID = "dnet:provenanceActions";

    private final static String COUNTER_GROUP = "Bulk Tagging";

    private String trust = "0.8";

    public OafProtos.Oaf enrichContext(final OafProtos.Oaf oaf, final CommunityConfiguration conf, final Mapper.Context context) {

        //context.getCounter(COUNTER_GROUP, "to enrich").increment(1);

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

        if(oaf.getDataInfo().getDeletedbyinference()){

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

            return null;

        }

        //context.getCounter(COUNTER_GROUP, "not deleted by inference").increment(1);

        final List<ResultProtos.Result.Context> contextList = oaf.getEntity().getResult().getMetadata().getContextList();

        if(contextList.size()>0){

            context.getCounter(COUNTER_GROUP, "exist context list").increment(1);

        }else{

            context.getCounter(COUNTER_GROUP, "not exist context list").increment(1);

        }

        //communities contains all the communities to be added as context for the result

        final Set<String> communities = new HashSet<>();

        oaf.getEntity().getResult().getMetadata().getSubjectList().stream()

                .map(subject -> subject.getValue())

                .filter(StringUtils::isNotBlank)

                .map(String::toLowerCase)

                .map(String::trim)

                .collect(Collectors.toCollection(HashSet::new))

                .forEach(s -> communities.addAll(conf.getCommunityForSubjectValue(s)));

        oaf.getEntity().getResult().getInstanceList()

                .stream()

                .map(i -> new Pair<>(i.getCollectedfrom().getKey(), i.getHostedby().getKey()))

                .flatMap(p -> Stream.of(p.getFst(), p.getSnd()))

                .map(s -> StringUtils.substringAfter(s, "|"))

                .collect(Collectors.toCollection(HashSet::new))

                .forEach(dsId -> communities.addAll(conf.getCommunityForDatasourceValue(dsId)));

        //TODO: add code for Zenodo Communities

        if(communities.isEmpty()){

            context.getCounter(COUNTER_GROUP, "list of communities empty").increment(1);

        }else{

            context.getCounter(COUNTER_GROUP, "list of communities has values!").increment(1);

        }

        final ResultProtos.Result.Metadata.Builder mBuilder = builder.getEntityBuilder().getResultBuilder().getMetadataBuilder();

        final Map<String, ResultProtos.Result.Context.Builder> cBuilders = Maps.newHashMap();

        mBuilder.getContextBuilderList().forEach(cBuilder -> {

            cBuilders.put(cBuilder.getId(), cBuilder);

        });

        for(String contextId:communities){

            final ResultProtos.Result.Context.Builder cBuilder = cBuilders.get(contextId);

            if (cBuilder != null) {

                if (!cBuilder.getDataInfoBuilderList().stream()

                        .map(di -> di.getInferenceprovenance())

                        .anyMatch(s -> DATA_INFO_TYPE.equals(s))) {

                    cBuilder.addDataInfo(buildDataInfo());

                    context.getCounter(COUNTER_GROUP, "add provenance").increment(1);

                } else {

                    context.getCounter(COUNTER_GROUP, "provenance already bulk tagged").increment(1);

                }

            } else {

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

                mBuilder.addContext(buildContext(contextId));

            }

        }

        return builder.build();

    }

    private ResultProtos.Result.Context buildContext(final String contextId) {

        return ResultProtos.Result.Context.newBuilder()

                .setId(contextId)

                .addDataInfo(buildDataInfo())

                .build();

    }

    private FieldTypeProtos.DataInfo buildDataInfo() {

        FieldTypeProtos.DataInfo.Builder builder = FieldTypeProtos.DataInfo.newBuilder()

                .setInferred(true)

                .setProvenanceaction(

                        FieldTypeProtos.Qualifier.newBuilder()

                                .setClassid(CLASS_ID)

                                .setClassname("Bulk Tagging for Communities")

                                .setSchemeid(SCHEMA_ID)

                                .setSchemename(SCHEMA_NAME))

                .setInferenceprovenance(DATA_INFO_TYPE)

                .setTrust(trust);

        return builder

                .build();

    }

    public void setTrust(String s) {

        trust = s;

    }

}

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

import java.io.IOException;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

public class ExportSimplifiedRecordsReducer extends Reducer<Text, Text, Text, Text> {

	private Text keyOut;

	@Override

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

		keyOut = new Text("");

	}

	@Override

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

		for(final Text v : values) {

			//keyOut.set(key.toString() + "@@@");

			context.write(keyOut, v);

		}

	}

}

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

import java.io.IOException;

import java.util.Map;

import java.util.Map.Entry;

import java.util.NavigableMap;

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.io.Text;

import com.google.common.base.Joiner;

import com.googlecode.protobuf.format.JsonFormat;

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

/**

 * Exports Oaf objects as their json serialization.

 *

 * @author claudio

 *

 */

public class ExportInformationSpaceMapper extends TableMapper<Text, Text> {

	/**

	 * logger.

	 */

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

	private static final String SEPARATOR = "@";

	private Text keyOut;

	private Text valueOut;

	@Override

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

		super.setup(context);

		keyOut = new Text();

		valueOut = new Text();

	}

	@Override

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

		try {

			byte[] rowKey = keyIn.copyBytes();

			Map<byte[], NavigableMap<byte[], byte[]>> row = value.getNoVersionMap();

			for (byte[] cf : row.keySet()) {

				for (Entry<byte[], byte[]> q : row.get(cf).entrySet()) {

					emit(rowKey, cf, q.getKey(), q.getValue(), context);

				}

			}

		} catch (final Throwable e) {

			log.error("error exporting the following record from HBase: " + value.toString(), e);

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

			throw new RuntimeException(e);

		}

	}

	private void emit(final byte[] rowKey, final byte[] cf, final byte[] q, final byte[] value, final Context context) throws IOException, InterruptedException {

		keyOut.set(Joiner.on(SEPARATOR).join(new String(rowKey), new String(cf), new String(q)));

		if ((value == null) || (value.length == 0)) {

			valueOut.set("");

		} else {

			valueOut.set(new JsonFormat().printToString(OafDecoder.decode(value).getOaf()));

		}

		context.write(keyOut, valueOut);

	}

}

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

import java.io.IOException;

import java.util.ArrayList;

import java.util.Collections;

import java.util.List;

import com.google.common.base.Function;

import com.google.common.collect.Iterables;

import com.google.common.collect.Lists;

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

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

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

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;

/**

 * Exports the result identifiers as json.

 *

 * @author claudio

 */

public class ExportResultIdentifiersMapper extends TableMapper<Text, Text> {

	/**

	 * logger.

	 */

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

	private static final String CF = "result";

	private Text keyOut;

	private Text valueOut;

	@Override

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

		super.setup(context);

		keyOut = new Text("");

		valueOut = new Text();

	}

	@Override

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

		try {

			final byte[] body = value.getValue(Bytes.toBytes(CF), DedupUtils.BODY_B);

			if (body == null) {

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

				return;

			}

			final OpenaireEntityId id = new OpenaireEntityId();

			final OafDecoder d = OafDecoder.decode(body);

			id.setDeleted(d.getOaf().getDataInfo().getDeletedbyinference());

			id.setId(d.getEntityId());

			final List<OafEntity> childrenList = d.getEntity().getChildrenList();

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

				final ArrayList<String> mergedIds = Lists.newArrayList(Iterables.transform(childrenList, new Function<OafEntity, String>() {

					@Override

					public String apply(final OafEntity oafEntity) {

						return oafEntity.getId();

					}

				}));

				Collections.sort(mergedIds);

				id.setMergedIds(mergedIds);

			}

			valueOut.set(id.toString());

			context.write(keyOut, valueOut);

		} catch (final Throwable e) {

			log.error("error exporting the following record from HBase: " + value.toString(), e);

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

			throw new RuntimeException(e);

		}

	}

}

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

import java.util.List;

import com.google.gson.Gson;

/**

 * Created by claudio on 27/01/16.

 */

public class OpenaireEntityId {

	private String id;

	private List<String> mergedIds;

	private boolean deleted;

	public OpenaireEntityId() {

	}

	public OpenaireEntityId(final String id, final List<String> mergedIds, final boolean deleted) {

		this.id = id;

		this.mergedIds = mergedIds;

		this.deleted = deleted;

	}

	public String getId() {

		return id;

	}

	public void setId(final String id) {

		this.id = id;

	}

	public List<String> getMergedIds() {

		return mergedIds;

	}

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

		this.mergedIds = mergedIds;

	}

	public boolean isDeleted() {

		return deleted;

	}

	public void setDeleted(final boolean deleted) {

		this.deleted = deleted;

	}

	@Override

	public String toString() {

		return new Gson().toJson(this);

	}

}

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

import java.io.IOException;

import eu.dnetlib.miscutils.functional.xml.ApplyXslt;

import org.apache.commons.codec.binary.Base64;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

public class ExportSimplifiedRecordsMapper extends Mapper<Text, Text, Text, Text> {

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

	private ApplyXslt recordSummarizer;

	private Text valueOut;

	private Text keyOut;

	@Override

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

		final String xslt = new String(Base64.decodeBase64(context.getConfiguration().get("xslt")));

		log.info("got xslt: \n" + xslt);

		recordSummarizer = new ApplyXslt(xslt);

		valueOut = new Text();

		keyOut = new Text("");

	}

	@Override

	protected void map(final Text key, final Text value, final Context context) throws IOException, InterruptedException {

		final String summary = recordSummarizer.evaluate(value.toString());

		if (StringUtils.isNotBlank(summary)) {

		    keyOut.set(StringUtils.substringAfter(key.toString(), "::"));

			valueOut.set(summary.replaceAll("\n","").replaceAll("\t",""));

			context.write(keyOut, valueOut);

		}

	}

}

package eu.dnetlib.data.mapreduce.hbase;

public enum VolatileColumnFamily {

	dedup, dedupPerson; // instance is here to remove the old protos

	public static boolean isVolatile(final String columnName) {

		try {

			return VolatileColumnFamily.valueOf(columnName) != null;

		} catch (final Throwable e) {

			return false;

		}

	}

}

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.TypeProtos.Type;

import eu.dnetlib.pace.config.DedupConfig;

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;

public class DedupDeleteRelMapper 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());

		deleteRels(rowkey, context, value, DedupUtils.getSimilarityCFBytes(type));

		deleteRels(rowkey, context, value, DedupUtils.getDedupCF_mergedInBytes(type));

		deleteRels(rowkey, context, value, DedupUtils.getDedupCF_mergesBytes(type));

	}

	private void deleteRels(final ImmutableBytesWritable rowkey, final Context context, final Result value, final byte[] cf)

			throws IOException, InterruptedException {

		final Map<byte[], byte[]> rels = value.getFamilyMap(cf);

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

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

			final Delete delete = new Delete(row);

			delete.setWriteToWAL(JobParams.WRITE_TO_WAL);

			delete.deleteFamily(cf);

			outKey.set(row);

			context.write(outKey, delete);

			context.getCounter(dedupConf.getWf().getEntityType(), new String(cf) + " deleted").increment(rels.size());

		}

	}

}

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

import java.io.IOException;

import java.nio.ByteBuffer;

import java.util.Map;

import com.google.protobuf.InvalidProtocolBufferException;

import eu.dnetlib.data.mapreduce.JobParams;

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

import eu.dnetlib.data.proto.DedupProtos.Dedup;

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

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

import eu.dnetlib.data.proto.OafProtos.OafRel.Builder;

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

import eu.dnetlib.data.transform.xml.AbstractDNetXsltFunctions;

import eu.dnetlib.pace.config.DedupConfig;

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 DedupFindRootsMapper extends TableMapper<ImmutableBytesWritable, Put> {

	private DedupConfig dedupConf;

	@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());

	}

	@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 ByteBuffer min = findMin(ByteBuffer.wrap(rowkey.get()), similarRels.keySet());

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

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

			// System.out.println("Found root: " + new String(root));

			emitDedupRel(context, DedupUtils.getDedupCF_mergedInBytes(type), row, root, buildRel(row, root, Dedup.RelName.isMergedIn));

			emitDedupRel(context, DedupUtils.getDedupCF_mergesBytes(type), root, row, buildRel(root, row, Dedup.RelName.merges));

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

			// marks the original body deleted

			emitBody(context, row, value.getValue(Bytes.toBytes(dedupConf.getWf().getEntityType()), DedupUtils.BODY_B));

		} else {

			context.getCounter(dedupConf.getWf().getEntityType(), "row not in similarity mesh").increment(1);

		}

	}

	private ByteBuffer findMin(ByteBuffer min, final Iterable<byte[]> keys) {

		for (final byte[] q : keys) {

			final ByteBuffer iq = ByteBuffer.wrap(q);

			if (min.compareTo(iq) > 0) {

				min = iq;

			}

		}

		return min;

	}

	private void emitBody(final Context context, final byte[] row, final byte[] body) throws InvalidProtocolBufferException, IOException, InterruptedException {

		if (body == null) {

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

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

			return;

		}

		final Oaf prototype = Oaf.parseFrom(body);

		if (prototype.getDataInfo().getDeletedbyinference()) {

			context.getCounter(dedupConf.getWf().getEntityType(), "bodies already deleted").increment(1);

		} else {

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

			oafRoot.getDataInfoBuilder().setDeletedbyinference(true).setInferred(true).setInferenceprovenance(dedupConf.getWf().getConfigurationId());

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

			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(dedupConf.getWf().getEntityType(), "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 Oaf oaf =

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

	}

	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);

	}

}

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

import java.io.StringWriter;

import java.util.List;

import java.util.Set;

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 org.apache.commons.lang.StringUtils;

/**

 * Created by claudio on 26/04/16.

 */

public class CsvSerialiser {

	private final static int MAX_FEATURES = 1000;

	private final static int MAX_ROWS = 5000;

	private int maxRows = MAX_ROWS;

	private int maxFeatures = MAX_FEATURES;

	public CsvSerialiser() {

	}

	public CsvSerialiser(int maxRows, int maxFeatures) {

		this.maxRows = maxRows;

		this.maxFeatures = maxFeatures;

	}

	public String asCSV(final List<CsvEntry> list) {

		final Set<String> features = Sets.newLinkedHashSet();

		for(CsvEntry e : Iterables.limit(list, maxRows)) {

			features.addAll(e.getFeatures());

		}

		final List<String> cappedFeatures = Lists.newLinkedList(Iterables.limit(features, maxFeatures));

		//context.getCounter("person", "features " + Iterables.size(cappedFeatures)).increment(1);

		final StringWriter csv = new StringWriter();

		csv.append("\"k\",");

		csv.append(Joiner.on(",").join(cappedFeatures));

		csv.append(",\"id\",\"name\",\"title\"\n");

		for(CsvEntry e : Iterables.limit(list, maxRows)) {

			boolean hasZero = false;

			boolean hasOne = false;

			final StringWriter line = new StringWriter();

			line.append(e.getKey()+",");

			for(String f : cappedFeatures) {

				if(e.getFeatures().contains(f)) {

					line.append("1,");

					hasOne = true;

				} else {

					line.append("0,");

					hasZero = true;

				}

			}

			line.append("\""+e.getId()+"\",");

			line.append("\""+e.getOriginalName()+"\",");

			line.append("\""+e.getTitle()+"\"");

			if (hasZero && hasOne) {

				csv.append(line.toString() + "\n");

			}

			//csv.append(StringUtils.substringBeforeLast(line.toString(), ",")  + "\n");

		}

		return csv.toString();

	}

}

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

import java.util.HashMap;

import java.util.Map.Entry;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.bson.BsonDocument;

import org.bson.BsonDocumentWrapper;

import org.bson.codecs.configuration.CodecRegistry;

import org.bson.conversions.Bson;

/**

 * Created by claudio on 07/03/16.

 */

public class SubjectsMap extends HashMap<String, Subjects> {

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

	public SubjectsMap mergeFrom(SubjectsMap sm) {

		if (sm != null) {

			for (Entry<String, Subjects> e : sm.entrySet()) {

				if (!this.containsKey(e.getKey())) {

					Subjects sub = new Subjects();

					sub.addAll(e.getValue());

					this.put(e.getKey(), sub);

				} else {

					for (String s : e.getValue()) {

						final Subjects subjects = this.get(e.getKey());

						subjects.add(s);

					}

				}

			}

		}

		return this;

	}

}

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

import java.io.IOException;

import java.util.List;

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

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

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

import eu.dnetlib.data.proto.ResultProtos;

import org.apache.commons.lang.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 OafDecoder decoder = OafDecoder.decode(body);

		final ResultProtos.Result result = decoder.getEntity().getResult();

		if (result.getMetadata().getResulttype().getClassid().equals("dataset")) {

			context.getCounter(RESULT, "dataset").increment(1);

			return;

		} else {

			context.getCounter(RESULT, "publication").increment(1);

		}

		if (result.getMetadata().getDescriptionCount() > MAX_DESCRIPTIONS) {

			context.getCounter(RESULT, "abstracts > " + MAX_DESCRIPTIONS).increment(1);

		} else {

			context.getCounter(RESULT, "abstracts: " + result.getMetadata().getDescriptionCount()).increment(1);

		}

		final List<StringField> descList = result.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.util.HashSet;

import org.bson.BsonDocument;

import org.bson.BsonDocumentWrapper;

import org.bson.codecs.configuration.CodecRegistry;

import org.bson.conversions.Bson;

/**

 * Created by claudio on 07/03/16.

 */

public class Subjects extends HashSet<String> {

}

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

import java.util.List;

import com.google.common.base.Splitter;

import org.apache.commons.lang.StringUtils;

import org.dom4j.Element;

/**

 * Created by claudio on 25/03/16.

 */

public class SubjectParser {

	public static final String REGEX_SUBJECT = "^(info:eu-repo)\\/(classification)\\/([a-zA-Z]*)\\/(.*)$";

	private static final int MIN_LENGTH = 5;

	public SubjectsMap parse(final org.dom4j.Document doc) {

		final List subjectNodes = doc.selectNodes("//*[local-name() = 'subject']");

		final SubjectsMap subjectMap = new SubjectsMap();

		for(int i = 0; i<subjectNodes.size(); i++) {

			final Element e = (Element) subjectNodes.get(i);

			final String subject = e.getText();

			final String type = guessType(subject);

			if (!subjectMap.containsKey(type)) {

				subjectMap.put(type, new Subjects());

			}

			if (StringUtils.isNotBlank(type)) {

				if ("keyword".equals(type)) {

					final Splitter splitter = Splitter.on(",").trimResults().omitEmptyStrings();

					for (String token : splitter.split(subject)) {

						final String value = token.replaceAll("[^a-zA-Z ]", "").toLowerCase();

						if (value.length() >= MIN_LENGTH) {

							subjectMap.get(type).add(value);

						}

					}

				} else {

					String token = subject.replaceFirst(REGEX_SUBJECT, "$4");

					if (StringUtils.isNotBlank(token)) {

						final String value = token.replaceAll("[^a-zA-Z ]", "").toLowerCase();

						if (value.length() >= MIN_LENGTH) {

							subjectMap.get(type).add(value);

						}

					}

				}

			}

		}

		return subjectMap;

	}

	private String guessType(final String subject) {

		if (subject.startsWith("info:eu-repo")) {

			final String s = subject.replaceAll(REGEX_SUBJECT, "$3");

			return s;

		} else {

			return "keyword";

		}

	}

}

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

import java.util.Set;

import com.google.common.collect.Sets;

import com.google.gson.Gson;

/**

 * Created by claudio on 20/04/16.

 */

public class CsvEntry {

	private String key;

	private Set<String> features = Sets.newLinkedHashSet();

	private String title;

	private String id;

	private String originalName;

	public CsvEntry() {

	}

	public CsvEntry(final String key, final Set<String> features) {

		this.key = key;

		this.features = features;

	}

	public CsvEntry(final Set<String> features) {

		this.features = features;

	}

	public void addFeature(final String f) {

		getFeatures().add(f);

	}

	public Set<String> getFeatures() {

		return features;

	}

	public void setFeatures(final Set<String> features) {

		this.features = features;

	}

	public static CsvEntry fromJson(final String json) {

		return new Gson().fromJson(json, CsvEntry.class);

	}

	public String getKey() {

		return key;

	}

	public void setKey(final String key) {

		this.key = key;

	}

	public String getTitle() {

		return title;

	}

	public void setTitle(final String title) {

		this.title = title;

	}

	public String getId() {

		return id;

	}

	public void setId(final String id) {

		this.id = id;

	}

	@Override

	public String toString() {

		return new Gson().toJson(this);

	}

	@Override

	public boolean equals(final Object o) {

		return (o instanceof CsvEntry) && ((CsvEntry) o).getFeatures().equals(getFeatures());

	}

	public String getOriginalName() {

		return originalName;

	}

	public void setOriginalName(final String originalName) {

		this.originalName = originalName;

	}

}

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

import java.io.IOException;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

/**

 * Created by claudio on 15/10/15.

 */

public class ConnectedComponentsMapper extends Mapper<Text, VertexWritable, Text, VertexWritable> {

	@Override

	protected void map(Text key, VertexWritable value, Context context) throws IOException, InterruptedException {

		context.write(value.getVertexId(), value);

	}

}

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.DedupProtos.Dedup;

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

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

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

import eu.dnetlib.data.proto.OafProtos.OafRel.Builder;

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 OafRel.Builder oafRef = DedupUtils.getDedup(dedupConf, new String(from), new String(to), relClass);

		final Oaf oaf = DedupUtils.buildRel(dedupConf, oafRef, 0.8).build();

		return oaf.toByteArray();

	}

}

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

import java.io.IOException;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

/**

 * Created by claudio on 14/10/15.

 */

public class MindistSearchMapper extends Mapper<Text, VertexWritable, Text, VertexWritable> {

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

	private boolean debug = false;

	@Override

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

		super.setup(context);

		debug = context.getConfiguration().getBoolean("mindist_DEBUG", false);

		log.info("debug mode: " + debug);

	}

	@Override

	protected void map(Text key, VertexWritable value, Context context) throws IOException, InterruptedException {

		emit(key, value, context);

		if (value.isActivated()) {

			VertexWritable vertex = new VertexWritable();

			for (Text edge : value.getEdges()) {

				if (!edge.toString().equals(value.getVertexId().toString())) {

					vertex.setVertexId(value.getVertexId());

					vertex.setEdges(null);

					emit(edge, vertex, context);

				}

			}

		}

	}

	private void emit(final Text key, final VertexWritable vertex, final Context context) throws IOException, InterruptedException {

		context.write(key, vertex);

		if (debug) {

			log.info(vertex.toJSON());

		}

	}

}

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

import java.io.IOException;

import org.apache.hadoop.hbase.KeyValue;

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;

/**

 * Created by claudio on 14/10/15.

 */

public class HBaseToSimilarityGraphMapper extends TableMapper<Text, VertexWritable> {

	@Override

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

		final VertexWritable vertex = new VertexWritable();