D-Net

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

import com.google.protobuf.InvalidProtocolBufferException;

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

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

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

import eu.dnetlib.data.mapreduce.hbase.lodExport.utils.Serializer;

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

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

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

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

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

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

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

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.log4j.Logger;

import java.io.IOException;

import java.text.ParseException;

import java.text.SimpleDateFormat;

import java.util.*;

import java.util.Map.Entry;

/**

 * Mapper Class that reads HBASE contents and prepares them for the StatsDB

 * export

 */

public class LodMapper extends TableMapper<Text, Text> {

	private static final String DATE_OF_TRANSFORMATION_PATTERN = "yyyy-MM-dd'T'HH:mm:ss";

	private static final String LAST_EXECUTION_DATE_PATTERN = "yyyy-MM-dd";

	private Logger log = Logger.getLogger(this.getClass());

	private EntityConfigTable entityConfigTable;

	private String lastExecutionDate = "";

	public static enum ENTITIES_COUNTER {

		RESULT, PROJECT, DATASOURCE, PERSON, ORGANIZATION, RESULT_NO_DATE, PROJECT_NO_DATE, DATASOURCE_NO_DATE,

		PERSON_NO_DATE, RGANIZATION_NO_DATE, DELETED_BY_INFERENCE, NOT_DELETED_BY_INFERENCE, TOTAL_ENTITIES,

		TOTAL_RELATIONS, UPDATED, NOT_UPDATED, NO_DATE

	}

	;

	private String DELIM;

	@Override

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

		loadEntityConfig(context);

		DELIM = context.getConfiguration().get("lod.delim");

		lastExecutionDate = context.getConfiguration().get("lod.lastExecutionDate");

	}

	@Override

	protected void map(final ImmutableBytesWritable keyIn, final Result result, final Context context)

			throws IOException {

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

		final Type type = decoder.getType();

		final Oaf oaf = UpdateMerger.mergeBodyUpdates(context, result.getFamilyMap(Bytes.toBytes(type.toString())));

		if (isValid(oaf)) {

			if (deletedByInference(oaf)) {

				context.getCounter(ENTITIES_COUNTER.DELETED_BY_INFERENCE).increment(1);

			} else {

				context.getCounter(ENTITIES_COUNTER.NOT_DELETED_BY_INFERENCE).increment(1);

			}

			context.getCounter(ENTITIES_COUNTER.TOTAL_ENTITIES).increment(1);

			try {

				emitProtos(context, result, oaf);

			} catch (ParseException e) {

				// TODO Auto-generated catch block

				e.printStackTrace();

			}

		}

	}

	private boolean isValid(Oaf oaf) {

		try {

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

				return true;

			}

		} catch (Exception e) {

			log.error("invalid proto", e);

		}

		return false;

	}

	private boolean isUpdated(Oaf oaf, Context context) throws IOException, ParseException {

		String dateOfTransformationString = "";

		try {

			SimpleDateFormat simpleDateFormat = new SimpleDateFormat(DATE_OF_TRANSFORMATION_PATTERN,

					Locale.getDefault());

			simpleDateFormat.setTimeZone(TimeZone.getTimeZone("UTC"));

			dateOfTransformationString = oaf.getEntity().getDateoftransformation();

			if (dateOfTransformationString == null || dateOfTransformationString.isEmpty()

					|| dateOfTransformationString.equals(" ")) {

				context.getCounter(ENTITIES_COUNTER.NO_DATE).increment(1);

				return true;

			}

			Date dateOfTransformation = simpleDateFormat.parse(dateOfTransformationString);

			SimpleDateFormat lastExecDateFormatter = new SimpleDateFormat(LAST_EXECUTION_DATE_PATTERN);

			simpleDateFormat.setTimeZone(TimeZone.getTimeZone("UTC"));

			Date lastExecDate = lastExecDateFormatter.parse(lastExecutionDate);

			if (lastExecDate.before(dateOfTransformation)) {

				context.getCounter(ENTITIES_COUNTER.UPDATED).increment(1);

				return true;

			}

		} catch (ParseException pe) {

			SimpleDateFormat simpleDateFormat = new SimpleDateFormat(LAST_EXECUTION_DATE_PATTERN, Locale.getDefault());

			simpleDateFormat.setTimeZone(TimeZone.getTimeZone("UTC"));

			dateOfTransformationString = oaf.getEntity().getDateoftransformation();

			if (dateOfTransformationString == null || dateOfTransformationString.isEmpty()

					|| dateOfTransformationString.equals(" ")) {

				context.getCounter(ENTITIES_COUNTER.NO_DATE).increment(1);

				return true;

			}

			Date dateOfTransformation = simpleDateFormat.parse(dateOfTransformationString);

			SimpleDateFormat lastExecDateFormatter = new SimpleDateFormat(LAST_EXECUTION_DATE_PATTERN);

			simpleDateFormat.setTimeZone(TimeZone.getTimeZone("UTC"));

			Date lastExecDate = lastExecDateFormatter.parse(lastExecutionDate);

			if (lastExecDate.before(dateOfTransformation)) {

				context.getCounter(ENTITIES_COUNTER.UPDATED).increment(1);

				return true;

			}

		} catch (Exception e) {

			log.error("invalid date " + dateOfTransformationString, e);

			throw new IOException(e);

		}

		context.getCounter(ENTITIES_COUNTER.NOT_UPDATED).increment(1);

		return false;

	}

	private void emitProtos(Context context, Result result, Oaf oaf) throws IOException, ParseException {

		Oaf.Builder oafBuilder = Oaf.newBuilder().setKind(oaf.getKind()).setDataInfo(oaf.getDataInfo());

		Type type = oaf.getEntity().getType();

		oafBuilder.setEntity(oaf.getEntity());

		// emit relation first so we can cache them to entity protos

		emitRelation(context, result, oaf, type, oafBuilder);

		if (isUpdated(oaf, context)) {

			emitEntity(context, oaf, oafBuilder);

		}

	}

	private void emitEntity(Context context, Oaf oaf, Oaf.Builder oafBuilder) {

		String serialized = Serializer.serialize(oafBuilder.build(), DELIM);

		if (serialized != null && !oaf.getEntity().getId().contains("dedup")) {

			try {

				context.getCounter(ENTITIES_COUNTER.valueOf(oaf.getEntity().getType().toString().toUpperCase()))

						.increment(1);

				// output entity types so we can have seperate files for each type

				Text TextKeyOut = new Text(oaf.getEntity().getType().toString());

				context.write((TextKeyOut), new Text(serialized));

				if (oaf.getEntity().getDateoftransformation().isEmpty()) {

					context.getCounter(

							ENTITIES_COUNTER.valueOf(oaf.getEntity().getType().toString().toUpperCase() + "_NO_DATE"))

							.increment(1);

//                    context.getCounter(ENTITIES_COUNTER.UPDATED).increment(1);

				}

//                else {

//                    context.getCounter(ENTITIES_COUNTER.NOT_UPDATED).increment(1);

//                }

			} catch (Exception e) {

				log.error("Error writing entity to M/R output", e);

			}

		}

	}

	// may have multiple relations per each field

	private void emitRelation(Context context, Result result, Oaf oaf, Type type, Oaf.Builder builder) {

		// derived relations; ResultLanguages, resultConcepts etc are

		// created here

		if (!oaf.getEntity().getId().contains("dedup")) {

			// Existing Hbase relations are generated here

			if (entityConfigTable.getDescriptors(type) != null && !entityConfigTable.getDescriptors(type).isEmpty()) {

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

					try {

						final Map<byte[], byte[]> columnMap = result

								.getFamilyMap(Bytes.toBytes(ld.getRelDescriptor().getIt()));

						List<OafRel> relOaf = decodeRelation(oaf, columnMap, ld);

						for (OafRel rel : relOaf) {

							builder.getEntityBuilder().addCachedRel(rel);

							try {

								// keep all relations to one file

								Text TextKeyOut = new Text("relations");

								String buff = Serializer.serialize(rel, DELIM);

								if (!buff.isEmpty() && !rel.getTarget().contains("dedup")) {

									context.write((TextKeyOut), new Text(buff));

									context.getCounter(ENTITIES_COUNTER.TOTAL_RELATIONS).increment(1);

								}

							} catch (Exception e) {

								log.error("Error while writing Relation Proto to M/R output", e);

							}

						}

						relOaf.clear();

					} catch (Exception e) {

						log.error("Error while decoding Relation Proto from HBase Body", e);

					}

				}

			}

		}

		Set<String> relationsList = new HashSet<String>();

		Serializer.extractRelations(oaf, DELIM, relationsList);

		for (String rel : relationsList) {

			try {

				Text TextKeyOut = new Text("relations");

				if (!oaf.getEntity().getId().contains("dedup")) {

					if (!rel.contains("dedup")) {

						context.write((TextKeyOut), new Text(rel));

						context.getCounter(ENTITIES_COUNTER.TOTAL_RELATIONS).increment(1);

					}

				} else {

					// for dedup entities write only dedup relationships: all the permutations

					// of children

					if (rel.contains("dedup")) {

						context.write((TextKeyOut), new Text(rel));

						context.getCounter(ENTITIES_COUNTER.TOTAL_RELATIONS).increment(1);

					}

				}

			} catch (Exception e) {

				log.error("Error writing relations to output : " + rel);

			}

		}

		relationsList.clear();

	}

	private ArrayList<OafRel> decodeRelation(final Oaf body, Map<byte[], byte[]> columnMap, final LinkDescriptor ld)

			throws IOException, InterruptedException {

		ArrayList<OafRel> rels = new ArrayList<OafRel>();

		if (hasData(columnMap)) {

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

				final Oaf decodedOaf = decodeProto(e.getValue());

				if (isValid(decodedOaf)) {

					OafRel.Builder relBuilder = OafRel.newBuilder(decodedOaf.getRel());

					// skip dedups

					if (ld.getRelDescriptor().getIt().contains(SubRelType.dedup.toString())

							&& isDedupSelf(relBuilder)) {

						// log.error("invalid protto", e);

					} else {

						OafRel oafRel = relBuilder.setRelType(ld.getRelDescriptor().getRelType()).build();

						relBuilder.setCachedTarget(body.getEntity());

						rels.add(oafRel);

					}

				}

			}

		}

		return rels;

	}

	private Oaf decodeProto(final byte[] body) {

		try {

			return Oaf.parseFrom(body);

		} catch (InvalidProtocolBufferException e) {

			log.error("Invalid Protos", e);

		}

		return null;

	}

	private void loadEntityConfig(Context context) throws InterruptedException {

		String indexConf = context.getConfiguration().get("index.conf");

		if (indexConf == null || indexConf.isEmpty()) {

			log.error("NULL ENTITY CONFIGURATION TABLE IN MAPPER  : ");

			throw new InterruptedException("Null enitity configuration in mappper");

		}

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

	}

	private boolean isDedupSelf(final OafRelOrBuilder rel) {

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

	}

	private boolean hasData(final Map<byte[], byte[]> columnMap) {

		return columnMap != null && !columnMap.isEmpty();

	}

	private boolean deletedByInference(final Oaf oaf) {

		return oaf.getDataInfo().getDeletedbyinference();

	}

	@Override

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

		super.cleanup(context);

	}

	public EntityConfigTable getEntityConfigTable() {

		return entityConfigTable;

	}

	public void setEntityConfigTable(EntityConfigTable entityConfigTable) {

		this.entityConfigTable = entityConfigTable;

	}

}