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.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,

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

    }

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

        }

    }

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

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

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

                return true;

            }

        } catch (Exception e) {

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

            throw new IOException(e);

        }

        return false;

    }

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

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

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

    }

}