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

import org.joda.time.format.DateTimeFormat;

import org.joda.time.format.DateTimeFormatter;

import java.io.IOException;

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 Logger log = Logger.getLogger(this.getClass());

    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,

        DELETED_BY_INFERENCE,

        NOT_DELETED_BY_INFERENCE,

        TOTAL_ENTITIES,

        TOTAL_RELATIONS,

        UPDATED,

        NOT_UPDATED

    }

    ;

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

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

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

            emitProtos(context, result, oaf);

        }

        //hack for timeout on long running mapper threads

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

        }

        return false;

    }

    //TODO : add is updated check in production !

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

        }

        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(), 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.UPDATED).increment(1);

                } else {

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

                }

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

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

        relationsList = null;

    }

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

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

        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;

    }

}