D-Net

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

import com.google.common.collect.Multimap;

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.proto.OafProtos.OafRel;

import eu.dnetlib.data.proto.OafProtos.OafRelOrBuilder;

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

import java.util.Map;

/**

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

        datasource,

        organization,

        result,

        person,

        project

    }

        resultProject,

        datasourceOrganization,

        organizationOrganization,

        resultResult,

        personPerson,

        personResult,

        projectOrganization,

        projectPerson,

        resultOrganization

    }

    private enum ERROR_COUNTERS {

        rottenRecords,

        rottenRelations

    }

    private int increment;

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

        }

        id += increment;

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

                emitProtos(context, result, oaf);

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

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

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

                } catch (Exception e) {

                }

            }

    // may have multiple relations per each field

        try {

            // writing out : personResult,ID,data

            // created here

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

                        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;

                    }

            }

        }

        // Existing Hbase relations are generated here

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

                            String buff = serializer.serialize(rel);

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

            }

        }

    }

                        RelDescriptor rd = ld.getRelDescriptor();

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

                    }

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

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

    }

            return Oaf.parseFrom(body);

        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;

    }

}