D-Net

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

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

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

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 String TABLE_MAP_PATH = "eu/dnetlib/data/mapreduce/hbase/statsExport/exportTables";

	private long threshold = 100000;

	@Override

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

		tableMappings = new Properties();

		InputStream file = ClassLoader.getSystemResourceAsStream(TABLE_MAP_PATH);

		tableMappings.load(file);

		file.close();

		InputStream in = ClassLoader.getSystemResourceAsStream("eu/dnetlib/data/mapreduce/hbase/statsExport/index.conf");

		byte[] b = new byte[in.available()];

		in.read(b);

		String data = new String(b);

		in.close();

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

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

		}

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

//		for (  Entry<Type, EntityConfig> ld : entityConfigTable.entrySet()) 

//		{	log.info("Table entityConfigTable " + ld.getKey() + ld.getValue().getLinks().entrySet());

//			

//		}

		serializer = new Serializer();

		serializer.setDelim(context.getConfiguration().get("mapred.output.delim"));

		serializer.setNullNum(context.getConfiguration().get("mapred.output.nullNum"));

		serializer.setNullString(context.getConfiguration().get("mapred.output.nullString"));

	}

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

			if (type != null) {

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

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

					emitProtos(context, result, oaf);

				}

			}

		}

	}

	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

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

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

			if (serialized != null) {

				try {

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

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

					incrementWrittenRecs();

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

				}

			}

		}

	}

	private void incrementWrittenRecs() {

		writtenRecs++;

		if (writtenRecs % threshold == 0) {

			log.info("recs: " + writtenRecs);

		}

	}

	// 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(Serializer.cleanId(rel.getKey()) + "," + serializer.getId(oaf));

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

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

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

							incrementWrittenRecs();

						}

					}

				}

			}

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

					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.write((TextKeyOut), new ImmutableBytesWritable(buff.getBytes()));

								incrementWrittenRecs();

							}

						} 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 (!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);

			e.printStackTrace();

		}

		return null;

	}

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

	}

}