D-Net

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

import eu.dnetlib.data.mapreduce.hbase.lodExport.utils.SourceParser;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

import org.apache.log4j.Logger;

import java.io.IOException;

/**

 * Mapper Class that reads HBASE contents and prepares them for the StatsDB

 * export

 */

public class SourceMapper extends Mapper<LongWritable, Text, Text, Text> {

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

    private SourceParser parser;

    public static enum ENTITIES_COUNTER {

        RESULT,

        PROJECT,

        DATASOURCE,

        PERSON,

        ORGANIZATION,

        SOURCE_ENTITIES,

    }

    @Override

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

//        lastExecutionDate = context.getConfiguration().get("lod.lastExecutionDate");

    }

    @Override

    protected void map(final LongWritable keyIn, final Text result, final Context context) throws IOException {

        try {

            //parse record accordingly

            String processedRecord = parser.parse(result.toString());

            //if record is valid according to our constaints

            if (processedRecord != null) {

                //here add id

                Text TextKeyOut = new Text("sourceId");

                context.write((TextKeyOut), new Text(processedRecord));

                context.getCounter(ENTITIES_COUNTER.SOURCE_ENTITIES).increment(1);

            }

        } catch (Exception e) {

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

        }

    }

    @Override

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

        super.cleanup(context);

    }

}