D-Net

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

import com.google.common.collect.ArrayListMultimap;

import com.google.common.collect.Multimap;

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

import eu.dnetlib.data.mapreduce.util.UpdateMerger;

import eu.dnetlib.data.proto.KindProtos;

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

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.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 StatsMapper extends TableMapper<Text, ImmutableBytesWritable> {

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

    //Index configuratios for OAFs

    private EntityConfigTable entityConfigTable;

    //Counters for produced entities. Visible in the JobTracker monitoring page.

    private enum STATS_COUNTERS {

        datasource,

        organization,

        result,

        person,

        project

    }

    private enum REL_COUNTERS {

        resultProject,

        datasourceOrganization,

        organizationOrganization,

        resultResult,

        personPerson,

        personResult,

        projectOrganization,

        projectPerson,

        resultOrganization

    }

    //Counter for corrupted records.

    private enum ERROR_COUNTERS {

        rottenRecords,

        rottenRelations

    }

    private Serializer serializer;

    private int id;

    private int increment;

    //Init class: Load Index config and mapping for DB tables.

    @Override

    protected void setup(Context context) {

        loadEntityConfig(context);

        this.serializer = new Serializer(context.getConfiguration().get("stats.delim"), context.getConfiguration().get("stats.enclChar"));

        id = context.getTaskAttemptID().getTaskID().getId();

        increment = context.getConfiguration().getInt("mapred.map.tasks", 0);

        if (increment == 0) {

            throw new IllegalArgumentException("mapred.map.tasks is zero");

        }

    }

    //Read HBASE table and decode Protos to OAF entities.

    @Override

    protected void map(final ImmutableBytesWritable keyIn, final Result result, final Context context) {

        // generating unique integer if for the entity.

        id += increment;

        try {

            OafRowKeyDecoder keyDecoder = OafRowKeyDecoder.decode(keyIn.copyBytes());

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

            if (isValid(oaf))

                emitProtos(context, result, oaf);

        } catch (Exception e) {

            log.error("Unable to parse proto in row: " + new String(keyIn.copyBytes()), e);

            context.getCounter(ERROR_COUNTERS.rottenRecords).increment(1);

        }

    }

    private boolean isValid(Oaf oaf) {

        return oaf != null && oaf.isInitialized() && !oaf.getDataInfo().getDeletedbyinference() && !oaf.getDataInfo().getInvisible();

    }

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

        Oaf.Builder oafBuilder = Oaf.newBuilder().setKind(oaf.getKind()).

                setDataInfo(oaf.getDataInfo()).setLastupdatetimestamp(oaf.getLastupdatetimestamp());

        oafBuilder.setEntity(oaf.getEntity());

        // emit relation first so we can cache them to entity protos

        emitRelation(context, result, oaf);

        emitEntity(context, oafBuilder.build());

    }

    private void emitEntity(Context context, Oaf oaf) {

        String serialized = serializer.serialize(oaf);

        if (serialized != null) {

                try {

                    Text TextKeyOut;

                    switch (oaf.getEntity().getType()) {

                        case project:

                        case datasource:

                        case organization:

                            TextKeyOut = new Text(oaf.getEntity().getType().toString() + "," + serializer.getId(oaf));

                            break;

                        case result:

                            default:

                            TextKeyOut = new Text(oaf.getEntity().getType().toString() + "," + id);

                            break;

                    }

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

                    context.getCounter(STATS_COUNTERS.valueOf(oaf.getEntity().getType().toString())).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) {

        try {

            // writing out : personResult,ID,data

            // derived relations; ResultLanguages, resultConcepts etc are

            // created here

            Multimap<String, String> relMap = ArrayListMultimap.create();

            serializer.extractRelations(oaf, relMap);

            if (!relMap.isEmpty()) {

                for (Entry<String, String> rel : relMap.entries()) {

                    Text TextKeyOut;

                    switch (oaf.getEntity().getType()) {

                        case project:

                        case datasource:

                        case organization:

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

                            break;

                        case result:

                        default:

                            TextKeyOut = new Text(rel.getKey() + "," + id);

                            break;

                    }

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

                }

            }

        } catch (Throwable e) {

            log.error("Error writing relation to M/R output", e);

        }

        // Existing Hbase relations are generated here

        if (entityConfigTable.getDescriptors(oaf.getEntity().getType()) != null) {

            for (LinkDescriptor ld : entityConfigTable.getDescriptors(oaf.getEntity().getType())) {

                try {

                    ArrayList<OafRel> oafRels = new ArrayList<OafRel>();

                    decodeRelation(oaf, context, result, ld, oafRels);

                    for (OafRel rel : oafRels) {

                        {

                            Text TextKeyOut = new Text(ld.getRelDescriptor().getRelType().toString() + "," + serializer.getId(rel));

                            String buff = serializer.serialize(rel);

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

                            context.getCounter(REL_COUNTERS.valueOf(rel.getRelType().toString())).increment(1);

                        }

                    }

                } catch (Throwable e) {

                    log.error("Error for record ", e);

                    context.getCounter(ERROR_COUNTERS.rottenRelations).increment(1);

                }

            }

        }

    }

    private void decodeRelation(final Oaf body, final Context context, Result result, final LinkDescriptor ld, List<OafRel> rels) {

        try {

            final Oaf.Builder oafBuilder = Oaf.newBuilder().setKind(KindProtos.Kind.relation);

            final Map<byte[], byte[]> columnMap = result.getFamilyMap(Bytes.toBytes(ld.getRelDescriptor().getIt()));

            if (hasData(columnMap)) {

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

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

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

                    if (ld.isSymmetric()) {

                        RelDescriptor rd = ld.getRelDescriptor();

                        relBuilder.setCachedTarget(body.getEntity()).setRelType(rd.getRelType()).setSubRelType(rd.getSubRelType());

                    }

                    OafRel oafRel = relBuilder.setChild(ld.isChild()).build();

                    rels.add(oafBuilder.setDataInfo(decodedOaf.getDataInfo()).setRel(oafRel).build().getRel());

                }

            }

        } catch (Throwable throwable) {

            log.error("Error Decoding relation for: " + body.getRel().getRelType() + " " + body.getEntity().getId() + " ", throwable);

            context.getCounter(ERROR_COUNTERS.rottenRelations).increment(1);

        }

    }

    private Oaf decodeProto(final byte[] body, Context context) {

        try {

            return Oaf.parseFrom(body);

        } catch (Exception e) {

            log.error(e);

            context.getCounter(ERROR_COUNTERS.rottenRecords).increment(1);

        }

        return null;

    }

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

    }

    public EntityConfigTable getEntityConfigTable() {

        return entityConfigTable;

    }

    public void setEntityConfigTable(EntityConfigTable entityConfigTable) {

        this.entityConfigTable = entityConfigTable;

    }

}