D-Net

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

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;

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.hadoop.mapreduce.Mapper.Context;

import org.apache.log4j.Logger;

import java.io.ByteArrayInputStream;

import java.io.IOException;

import java.io.InputStream;

import java.util.*;

import java.util.Map.Entry;

/**

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

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

        {

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

            if (!deletedByInference(oaf)) {

                emitProtos(context, result, oaf);

            }

        } 

    }

    private boolean isValid(Oaf oaf) {

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

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

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

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

                            }

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

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

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

                if (isValid(decodedOaf) && !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) {

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

    }

}