D-Net

package eu.dnetlib.data.mapreduce.hbase.broker.model;

import java.util.Date;

import java.util.Map;

import com.google.gson.Gson;

import com.google.gson.GsonBuilder;

// This class represents the ElasticSearch Entity

public class Event {

	public static final String DATE_FORMAT = "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'";

	private String eventId;

	private String producerId;

	private String topic;

	private String payload;

	private Date creationDate;

	private Date expiryDate;

	private boolean instantMessage;

	private Map<String, Object> map;

	public Event() {

	}

	public Event(final String producerId, final String eventId, final String topic, final String payload,

			final Date creationDate, final Date expiryDate, final boolean instantMessage,

			final Map<String, Object> map) {

		this.producerId = producerId;

		this.eventId = eventId;

		this.topic = topic;

		this.payload = payload;

		this.creationDate = creationDate;

		this.expiryDate = expiryDate;

		this.instantMessage = instantMessage;

		this.map = map;

	}

	public String getProducerId() {

		return this.producerId;

	}

	public void setProducerId(final String producerId) {

		this.producerId = producerId;

	}

	public String getEventId() {

		return this.eventId;

	}

	public void setEventId(final String eventId) {

		this.eventId = eventId;

	}

	public String getTopic() {

		return this.topic;

	}

	public void setTopic(final String topic) {

		this.topic = topic;

	}

	public String getPayload() {

		return this.payload;

	}

	public void setPayload(final String payload) {

		this.payload = payload;

	}

	public Date getCreationDate() {

		return this.creationDate;

	}

	public void setCreationDate(final Date creationDate) {

		this.creationDate = creationDate;

	}

	public Date getExpiryDate() {

		return this.expiryDate;

	}

	public void setExpiryDate(final Date expiryDate) {

		this.expiryDate = expiryDate;

	}

	public boolean isInstantMessage() {

		return this.instantMessage;

	}

	public void setInstantMessage(final boolean instantMessage) {

		this.instantMessage = instantMessage;

	}

	public Map<String, Object> getMap() {

		return this.map;

	}

	public void setMap(final Map<String, Object> map) {

		this.map = map;

	}

	public String asJson() {

		final Gson gson = new GsonBuilder().setDateFormat(DATE_FORMAT).create();

		return gson.toJson(this);

	}

}

package eu.dnetlib.data.mapreduce.hbase.broker.model;

import java.util.Date;

import java.util.stream.Collectors;

import org.apache.commons.codec.digest.DigestUtils;

import org.apache.hadoop.mapreduce.Mapper;

import org.apache.hadoop.mapreduce.Reducer;

//Utility class, it perfoms the conversion msg -> ES entinty and provides some and manages M/R logs

public class EventWrapper {

	private final EventMessage event;

	private final String highlightValue;

	private final String counterGroupName;

	private final String counterName;

	public static EventWrapper newInstance(final EventMessage event, final String highlightValue,

			final String counterGroupName, final String counterName) {

		return new EventWrapper(event, highlightValue, counterGroupName, counterName);

	}

	private EventWrapper(EventMessage event, String highlightValue, String counterGroupName, String counterName) {

		this.event = event;

		this.highlightValue = highlightValue;

		this.counterGroupName = counterGroupName;

		this.counterName = counterName;

	}

	public Event asBrokerEvent() {

		final Date now = new Date();

		final Event res = new Event();

		final String eventId = calculateEventId(event.getTopic(),

				event.getMap().get("target_publication_id").toString(), highlightValue);

		res.setEventId(eventId);

		res.setProducerId(event.getProducerId());

		res.setPayload(event.getPayload());

		res.setMap(event.getMap().entrySet().stream().filter(e -> e.getValue().asObject() != null)

				.collect(Collectors.toMap(e -> e.getKey(), e -> e.getValue().asObject())));

		res.setTopic(event.getTopic());

		res.setCreationDate(now);

		res.setExpiryDate(calculateExpiryDate(now, event.getTthDays()));

		res.setInstantMessage(event.getTthDays() == 0);

		return res;

	}

	private String calculateEventId(final String topic, final String publicationId, final String value) {

		return "event-" + DigestUtils.md5Hex(topic).substring(0, 6) + "-"

				+ DigestUtils.md5Hex(publicationId).substring(0, 8) + "-" + DigestUtils.md5Hex(value).substring(0, 8);

	}

	private Date calculateExpiryDate(final Date now, final int ttlDays) {

		if (ttlDays < 0) {

			return null;

		} else if (ttlDays == 0) {

			return now;

		} else {

			return new Date(now.getTime() + ttlDays * 24 * 60 * 60 * 1000);

		}

	}

	public void incrementCounter(Mapper<?, ?, ?, ?>.Context context) {

		context.getCounter(counterGroupName, counterName).increment(1);

	}

	public void incrementCounter(Reducer<?, ?, ?, ?>.Context context) {

		context.getCounter(counterGroupName, counterName).increment(1);

	}

}

package eu.dnetlib.data.mapreduce.hbase.broker.model;

import java.util.Map;

import com.google.gson.Gson;

/**

 * Created by claudio on 08/07/16.

 */

// This class represents the Event message as required by the Original API

// (rabbitMQ)

public class EventMessage {

	public static final int TTH_INFINITE = -1;

	private String producerId;

	private String topic;

	private String payload;

	private int tthDays = TTH_INFINITE;

	private Map<String, MapValue> map;

	public EventMessage() {

	}

	public EventMessage(final String producerId, final Topic topic, final String payload, final int tthDays,

			final Map<String, MapValue> map) {

		this.producerId = producerId;

		this.topic = topic.getValue();

		this.payload = payload;

		this.tthDays = tthDays;

		this.map = map;

	}

	public String getProducerId() {

		return this.producerId;

	}

	public void setProducerId(final String producerId) {

		this.producerId = producerId;

	}

	public String getTopic() {

		return this.topic;

	}

	public void setTopic(final String topic) {

		this.topic = topic;

	}

	public String getPayload() {

		return this.payload;

	}

	public void setPayload(final String payload) {

		this.payload = payload;

	}

	public Map<String, MapValue> getMap() {

		return this.map;

	}

	public void setMap(final Map<String, MapValue> map) {

		this.map = map;

	}

	public boolean hasNulls() {

		return (this.producerId == null) || (this.topic == null) || (this.payload == null) || (this.map == null);

	}

	public int getTthDays() {

		return this.tthDays;

	}

	public void setTthDays(final int tthDays) {

		this.tthDays = tthDays;

	}

	@Override

	public String toString() {

		return new Gson().toJson(this);

	}

}

package eu.dnetlib.data.mapreduce.hbase.broker.model;

import java.util.List;

import java.util.stream.Collectors;

import eu.dnetlib.data.mapreduce.hbase.broker.mapping.DateParser;

import org.apache.commons.lang3.BooleanUtils;

import org.apache.commons.lang3.StringUtils;

import org.apache.commons.lang3.math.NumberUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

public class MapValue {

	private MapValueType type = MapValueType.STRING;

	private Object value = "";

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

	public MapValue() {}

	public MapValue(final MapValueType type, final Object value) {

		this.type = type;

		this.value = value;

	}

	public MapValueType getType() {

		return this.type;

	}

	public void setType(final MapValueType type) {

		this.type = type;

	}

	public Object getValue() {

		return this.value;

	}

	public void setValue(final Object value) {

		this.value = value;

	}

	public Object asObject() {

		if (this.value == null) {

			log.warn("Object is NULL");

			return null;

		}

		try {

			switch (this.type) {

			case STRING:

			case INTEGER:

			case FLOAT:

			case BOOLEAN:

			case DATE:

				return asSimpleObject(getType(), getValue().toString());

			case LIST_STRING:

				return ((List<?>) getValue()).stream().map(o -> asSimpleObject(MapValueType.STRING, o.toString())).collect(Collectors.toList());

			case LIST_INTEGER:

				return ((List<?>) getValue()).stream().map(o -> asSimpleObject(MapValueType.INTEGER, o.toString())).collect(Collectors.toList());

			case LIST_FLOAT:

				return ((List<?>) getValue()).stream().map(o -> asSimpleObject(MapValueType.FLOAT, o.toString())).collect(Collectors.toList());

			case LIST_BOOLEAN:

				return ((List<?>) getValue()).stream().map(o -> asSimpleObject(MapValueType.BOOLEAN, o.toString())).collect(Collectors.toList());

			case LIST_DATE:

				return ((List<?>) getValue()).stream().map(o -> asSimpleObject(MapValueType.DATE, o.toString())).collect(Collectors.toList());

			default:

				log.warn("Invalid type: " + this.type);

				return null;

			}

		} catch (final Exception e) {

			log.warn("Error parsing value: " + this);

			return null;

		}

	}

	private Object asSimpleObject(final MapValueType type, final String s) {

		try {

			switch (type) {

			case STRING:

				return s.toString();

			case INTEGER:

				return s.contains(".") ? NumberUtils.toLong(StringUtils.substringBefore(s, "."), 0) : NumberUtils.toLong(s, 0);

			case FLOAT:

				return NumberUtils.toDouble(s, 0);

			case BOOLEAN:

				return BooleanUtils.toBoolean(s);

			case DATE:

				return DateParser.parse(s);

			default:

				log.warn("Unmamaged type: " + type);

				return null;

			}

		} catch (final Exception e) {

			log.warn("Error parsing value: " + s + " - type: " + type);

			return null;

		}

	}

	@Override

	public String toString() {

		return String.format("[ type; %s, value: %s ]", getType(), getValue());

	}

	}

package eu.dnetlib.data.mapreduce.hbase.broker.model;

/**

 * Created by claudio on 02/09/16.

 */

public enum MapValueType {

	STRING, INTEGER, FLOAT, DATE, BOOLEAN, LIST_STRING, LIST_INTEGER, LIST_FLOAT, LIST_DATE, LIST_BOOLEAN

}

package eu.dnetlib.data.mapreduce.hbase.broker.model;

public enum Topic {

	// ENRICHMENT MISSING

	ENRICH_MISSING_ARTICLE("ENRICH/MISSING/ARTICLE"),

	// ENRICHMENT MORE

	ENRICH_MORE_ARTICLE("ENRICH/MORE/ARTICLE");

	Topic(final String value) {

		this.value = value;

	}

	protected String value;

	public String getValue() {

		return this.value;

	}

}

package eu.dnetlib.data.mapreduce.hbase.broker.enrich;

import java.io.IOException;

import java.util.List;

import java.util.Objects;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

import eu.dnetlib.data.mapreduce.hbase.broker.model.EventWrapper;

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

	private static final int LIMIT = 1000;

	private Text tKey = new Text("");

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

	@Override

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

		super.setup(context);

		System.out.println("LIMIT: " + LIMIT);

	}

	@Override

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

			throws IOException, InterruptedException {

		// TODO Auto-generated method stub

		super.reduce(key, values, context);

	}

	private void emit(final List<EventWrapper> eventWrappers, final Context context) {

		eventWrappers.stream().filter(Objects::nonNull).forEach(ew -> emit(ew, context));

	}

	private void emit(final EventWrapper eventWrapper, final Context context) {

		try {

			final Text valueout = new Text(eventWrapper.asBrokerEvent().asJson());

			context.write(tKey, valueout);

			eventWrapper.incrementCounter(context);

		} catch (Exception e) {

			throw new RuntimeException(e);

		}

	}

}

package eu.dnetlib.data.mapreduce.hbase.broker.enrich;

import java.io.IOException;

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;

public class DliArticleEnrichmentMapper extends TableMapper<ImmutableBytesWritable, ImmutableBytesWritable> {

	protected ImmutableBytesWritable outValue;

	protected ImmutableBytesWritable outKey;

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

	@Override

	protected void setup(final Context context) {

		outKey = new ImmutableBytesWritable();

		outValue = new ImmutableBytesWritable();

	}

	@Override

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

			throws IOException, InterruptedException {

		// TODO Auto-generated method stub

		// DNGF d =

		super.map(key, value, context);

	}

	private void emit(final Context context, final byte[] key, final byte[] value, final String entityType) {

		outKey.set(key);

		outValue.set(value);

		try {

			context.write(outKey, outValue);

			context.getCounter("Broker Enrichment DLI articles", "entity type: " + entityType).increment(1);

		} catch (Exception e) {

			throw new IllegalArgumentException(e);

		}

	}

}

package eu.dnetlib.data.mapreduce.hbase.broker.mapping;

import eu.dnetlib.broker.objects.OpenAireEventPayload;

public class HighlightFactory {

	public static OpenAireEventPayload highlightArticle(final OpenAireEventPayload p, final String article) {

		// p.getHighlight().setInstances(mapInstances(instances));

		return p;

	}

}

package eu.dnetlib.data.mapreduce.hbase.broker.mapping;

import java.text.ParseException;

import org.apache.commons.lang3.time.DateUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

public class DateParser {

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

	public static Long parse(final String s) {

		if (s == null) {

			log.warn("Date is NULL");

			return null;

		}

		try {

			return DateUtils.parseDate(s, "yyyy-MM-dd", "yyyy-MM-dd HH:mm:ss").getTime();

		} catch (final ParseException e) {

			log.warn("Invalid Date: " + s);

			return null;

		}

	}

}

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

	xmlns="http://maven.apache.org/POM/4.0.0"

	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-openaire-broker-common</artifactId>

			<version>[1.0.0,2.0.0)</version>

		<!-- <dependency> <groupId>org.json</groupId> <artifactId>json</artifactId> 

			<version>20160212</version> <scope>test</scope> </dependency> -->