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

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

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 org.joda.time.DateTime;

import org.joda.time.format.DateTimeFormat;

import org.joda.time.format.DateTimeFormatter;

import java.io.IOException;

import java.util.ArrayList;

import java.util.List;

import java.util.Map;

import java.util.Map.Entry;

/**

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

 * export

 */

public class LodMapper extends TableMapper<Text, ImmutableBytesWritable> {

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

    private NewSerializer serializer;

    private EntityConfigTable entityConfigTable;

	//timeout: ~5 minutes, here in milliseconds

	private long threshold = 300000;

	private String lastExecutionDate = "";

	private long timestamp;

    public static enum ENTITIES_COUNTER {

        RESULT,

        PROJECT,

        DATASOURCE,

        PERSON,

        ORGANIZATION

    };

    @Override

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

		loadEntityConfig(context);

		serializer = new NewSerializer(context.getConfiguration().get("lod.delim"), context.getConfiguration().get("lod.seperator"));

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

		timestamp = System.currentTimeMillis();

    }

    @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)) {

            // TODO set this only when the configuration file has include dups

            // to fals for Results

            // or else were gonna get deleted by inference entries

             if (!deletedByInference(oaf))

			// entityConfigTable.includeDuplicates(type)) {

			//TODO eri: here allow entities deletedbyinfer

			{

				emitProtos(context, result, oaf);

			}

		}

		if (System.currentTimeMillis() - timestamp >= threshold) {

			context.progress();

			timestamp = System.currentTimeMillis();

		}

	}

    private boolean isValid(Oaf oaf) {

        try {

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

                return true;

            }

        } catch (Exception e) {

            // log.error("OAF NOT INITIALIZED ");

        }

        return false;

    }

    private boolean isUpdated(Oaf oaf) {

        try {

            DateTimeFormatter formatter = DateTimeFormat.forPattern("yyyy-MM-dd");

            //TODO here set data of trasformation

            String dateOfTransformationString = oaf.getEntity().getDateofcollection();

            if (dateOfTransformationString == ""|| dateOfTransformationString.isEmpty()|| dateOfTransformationString == null) return true;

            if (lastExecutionDate == "" || lastExecutionDate.isEmpty()||lastExecutionDate == null) return true;

            DateTime dt = formatter.parseDateTime(dateOfTransformationString);

            DateTime de = formatter.parseDateTime(lastExecutionDate);

            if (lastExecutionDate == "" || lastExecutionDate == null || dateOfTransformationString == "") return true;

            if (de.isBefore(dt)) 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());

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

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

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

            try {

                Text TextKeyOut = new Text("entities");

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

                //counter

                context.getCounter(type).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")&&!type.equals(Type.person)) {

            // 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 {

                                Text TextKeyOut = new Text("relations");

                                String buff = serializer.serialize(rel);

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

                    }

                }

            }

        }

        List<String> relationsList = serializer.extractRelations(oaf);

        for (String rel : relationsList) {

            try {

                Text TextKeyOut = new Text("relations");

                if (oaf.getEntity().getId().startsWith("dedup")) {

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

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

                    }

                } else {

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

                }

            } catch (Exception e) {

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

            }

        }

    }

    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 (hasData(columnMap)) {

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

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

                if (isValid(decodedOaf)) {

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

                    // skip dedups

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

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

                        //		relBuilder.setCachedTarget(body.getEntity());

                        //	rels.put("dedups", oafRel);

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

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

        }

        return null;

    }

    private void loadEntityConfig(Context context) {

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

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

    }

}