D-Net

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

import com.google.common.collect.ArrayListMultimap;

import com.google.common.collect.Multimap;

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

import eu.dnetlib.data.mapreduce.hbase.statsExport.utils.Serializer;

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

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

import java.io.IOException;

import java.io.InputStream;

import java.util.ArrayList;

import java.util.List;

import java.util.Map;

import java.util.Map.Entry;

import java.util.Properties;

/**

 * 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

    }

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

    @Override

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

        loadEntityConfig(context);

    }

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

    @Override

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

        Oaf oaf = null;

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

        try {

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

        } catch (Exception e) {

            /*

            System.err.println("Unable to parse proto in row: " + oaf.getEntity().getType().toString() + keyDecoder.getKey());

            log.error("Unable to parse proto in row: " + oaf.getEntity().getType().toString() + keyDecoder.getKey(), e);

            */

            System.err.println("Unable to parse proto in row: " + keyDecoder.getKey());

            log.error("Unable to parse proto in row: " + keyDecoder.getKey(), e);

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

        }

        if (isValid(oaf)) {

            emitProtos(context, result, oaf);

        }

    }

    private boolean isValid(Oaf oaf) {

        if (oaf != null && oaf.isInitialized() && !deletedByInference(oaf)) return true;

        return false;

    }

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

        emitEntity(context, oafBuilder.build());

    }

    private void emitEntity(Context context, Oaf oaf) {

        String serialized = Serializer.serialize(oaf, context.getConfiguration().get("stats.delim"), context.getConfiguration().get("stats.enclChar"));

        if (serialized != null) {

                try {

                    Text TextKeyOut = new Text(oaf.getEntity().getType().toString() + "," + Serializer.getId(oaf, context.getConfiguration().get("stats.delim"), context.getConfiguration().get("stats.enclChar")));

                    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, Oaf.Builder builder) {

        try {

            // writing out : personResult,ID,data

            // derived relations; ResultLanguages, resultConcepts etc are

            // created here

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

            Serializer.extractRelations(oaf, context.getConfiguration().get("stats.delim"), context.getConfiguration().get("stats.enclChar"), relMap);

            if (!relMap.isEmpty()) {

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

                    Text TextKeyOut = new Text(rel.getKey() + "," + Serializer.getId(oaf, context.getConfiguration().get("stats.delim"), context.getConfiguration().get("stats.enclChar")));

                    //TODO here output

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

                        {

                        //  builder.getEntityBuilder().addCachedRel(rel);

                            Text TextKeyOut = new Text(ld.getRelDescriptor().getRelType().toString()

                                    + "," + Serializer.getId(rel, context.getConfiguration().get("stats.delim"),

                                    context.getConfiguration().get("stats.enclChar")));

                            String buff = Serializer.serialize(rel, context.getConfiguration().get("stats.delim"), context.getConfiguration().get("stats.enclChar"));

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

    }

    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;

    }

}