D-Net

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

import java.io.ByteArrayInputStream;

import java.io.IOException;

import java.io.InputStream;

import java.util.ArrayList;

import java.util.HashMap;

import java.util.List;

import java.util.Map;

import java.util.Map.Entry;

import java.util.Properties;

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 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.statsExport.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;

/**

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

 * export

 */

public class StatsMapper extends TableMapper<Text, ImmutableBytesWritable> {

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

	private Serializer serializer;

	private EntityConfigTable entityConfigTable;

	private Properties tableMappings;

	private long writtenRecs = 0;

	private long threshold = 100000;

	public static enum STATS_COUNTER {

		INPUT_RELATIONS, WRITTEN_RELATIONS, OUTPUT_DERIVED_RELATIONS, INPUT_RECS, OUTPUT_RECS, ROTTEN_RECORDS, WRITTEN_ENTITIES, VALID_RECS, INPUT_ENTITIES

	};

	public static enum PROTOS_COUNTER {

		RESULT, PROJECT, DATASOURCE, ORGANIZATION, DATASOURCEORGANIZATION, DATASOURCETOPIC, DATASOURCELANGUAGE, PROJECTORGANIZATION, RESULTCLAIM, RESULTCLASSIFICATION, RESULTCONCEPT, RESULTLANGUAGE, RESULTORGANIZATION, RESULTRESULT, RESULTPROJECT, RESULTTOPIC, RESULTDATASOURCE

	};

	@Override

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

		loadTableMap(context);

		loadEntityConfig(context);

		serializer = new Serializer();

		serializer.setDELIM(context.getConfiguration().get("stats.delim"));

		serializer.setNULL_NUM(context.getConfiguration().get("stats.nullNum"));

		serializer.setNULL_STRING(context.getConfiguration().get("stats.nullString"));

		serializer.setENCLOSED(context.getConfiguration().get("stats.enclChar"));

	}

	@Override

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

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

		context.getCounter(STATS_COUNTER.INPUT_RECS).increment(1);

		final Type type = decoder.getType();

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

		if (isValid(oaf, context))

		{

			context.getCounter(STATS_COUNTER.VALID_RECS).increment(1);

			if (!deletedByInference(oaf) || entityConfigTable.includeDuplicates(type)) {

				emitProtos(context, result, oaf);

			}

		} else

		{

			context.getCounter(STATS_COUNTER.ROTTEN_RECORDS).increment(1);

		}

	}

	private boolean isValid(Oaf oaf, Context context) {

		try {

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

				return true;

			}

		} catch (Exception e) {

			log.error("OAF NOT INITIALIZED ");

		}

		return false;

	}

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

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

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

		emitEntity(context, oaf, type, oafBuilder);

	}

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

		// ONLY EMIT ENTITIES THAT WE WANT TO IMPORT IN STATS DB

		context.getCounter(STATS_COUNTER.INPUT_ENTITIES).increment(1);

		if (tableMappings.containsKey(type.toString())) {

			String serialized = serializer.serialize(oafBuilder.build());

			if (serialized != null) {

				try {

					Text TextKeyOut = new Text(type + "," + serializer.getId(oaf));

					context.getCounter(STATS_COUNTER.WRITTEN_ENTITIES).increment(1);

					context.getCounter(PROTOS_COUNTER.valueOf(type.toString().toUpperCase())).increment(1);

					context.write((TextKeyOut), new ImmutableBytesWritable(serialized.getBytes()));

				} catch (IOException e) {

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

				} catch (InterruptedException e) {

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

				}

			}

		}

	}

	// may have multiple relations per each field

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

		try {

			// derived relations; ResultLanguages, resultConcepts etc are

			// created here

			HashMap<String, List<String>> relMap = serializer.extractRelations(oaf);

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

				for (Entry<String, List<String>> rel : relMap.entrySet()) {

					Text TextKeyOut = new Text(rel.getKey() + "," + serializer.getId(oaf));

					for (String relVals : rel.getValue()) {

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

							context.getCounter(STATS_COUNTER.OUTPUT_DERIVED_RELATIONS).increment(1);

							context.write((TextKeyOut), new ImmutableBytesWritable(relVals.getBytes()));

						}

					}

				}

			}

		} catch (Exception e) {

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

		}

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

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

					context.getCounter(STATS_COUNTER.INPUT_RELATIONS).increment(1);

					for (OafRel rel : relOaf) {

						builder.getEntityBuilder().addCachedRel(rel);

						try {

							// TODO skip tables what we dont want to import (

							// like personresults)

							if (tableMappings.containsKey(rel.getRelType().toString()))

							{

								Text TextKeyOut = new Text(ld.getRelDescriptor().getRelType().toString() + "," + serializer.getId(rel));

								String buff = serializer.serialize(rel);

								context.getCounter(STATS_COUNTER.WRITTEN_RELATIONS).increment(1);

								// context.getCounter(PROTOS_COUNTER.valueOf(ld.getRelDescriptor().getRelType().toString().toUpperCase())).increment(1);

								context.write((TextKeyOut), new ImmutableBytesWritable(buff.getBytes()));

							}

						} catch (Exception e) {

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

						}

					}

				} catch (Exception e) {

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

				}

			}

		}

	}

	private ArrayList<OafRel> decodeRelation(final Oaf body, final Context context, Map<byte[], byte[]> columnMap, final LinkDescriptor ld) throws IOException, InterruptedException {

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

		if (isValid(columnMap)) {

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

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

				if (decodedOaf != null && !deletedByInference(decodedOaf)) {

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

					// skip dedups

					if (ld.getRelDescriptor().getIt().contains(SubRelType.dedup.toString()) && isDedupSelf(relBuilder)) {

						log.info("avoid to emit dedup self: " + relBuilder.getSource());

					}

					else {

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

						relBuilder.setCachedTarget(body.getEntity());

						rels.add(oafRel);

					}

				}

			}

		}

		return rels;

	}

	private Oaf decodeProto(final Context context, final byte[] body) {

		try {

			return Oaf.parseFrom(body);

		} catch (InvalidProtocolBufferException e) {

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

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

		}

		return null;

	}

	private void loadTableMap(Context context) throws IOException {

		tableMappings = new Properties();

		String tables = context.getConfiguration().get("stats.dbTablesMap");

		if (tables == null) {

			log.error("NULL TABLE MAP CONFIG  IN MAPPER  : ");

		}

		tables = tables.replaceAll(",", "\n");

		InputStream stream = new ByteArrayInputStream(tables.getBytes());

		tableMappings.load(stream);

		stream.close();

	}

	private void loadEntityConfig(Context context) {

		String indexConf = context.getConfiguration().get("stats.indexConf");

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

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

		}

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

	}

	private boolean isDedupSelf(final OafRelOrBuilder rel) {

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

	}

	private boolean isValid(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 Serializer getSerializer() {

		return serializer;

	}

	public void setSerializer(Serializer serializer) {

		this.serializer = serializer;

	}

	public EntityConfigTable getEntityConfigTable() {

		return entityConfigTable;

	}

	public void setEntityConfigTable(EntityConfigTable entityConfigTable) {

		this.entityConfigTable = entityConfigTable;

	}

	public Properties getTableMappings() {

		return tableMappings;

	}

	public void setTableMappings(Properties tableMappings) {

		this.tableMappings = tableMappings;

	}

	public long getWrittenRecs() {

		return writtenRecs;

	}

	public void setWrittenRecs(long writtenRecs) {

		this.writtenRecs = writtenRecs;

	}

	public long getThreshold() {

		return threshold;

	}

	public void setThreshold(long threshold) {

		this.threshold = threshold;

	}

}