D-Net

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

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

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;

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

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

    private LodConfiguration lodConfiguration;

    private final String SOURCE_LABEL = "http://www.eurocris.org/ontologies/cerif/1.3#name";

    private final String TARGET_LABEL = "http://www.w3.org/2000/01/rdf-schema#label";

    public static enum STOPWORD_COUNTERS {

        STOPWORD_RECORDS, WORDS

    }

    @Override

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

        lodConfiguration = new LodConfiguration();

    }

    @Override

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

        try {

            context.getCounter(STOPWORD_COUNTERS.STOPWORD_RECORDS).increment(1);

            String recordTitle = getRecordTitle(result.toString());

            if (recordTitle != null) {

                for (String titleWord : recordTitle.split(" ")) {

                    context.write(new Text(titleWord), new Text("1"));

                    context.getCounter(STOPWORD_COUNTERS.WORDS).increment(1);

                }

            }

        } catch (Exception e) {

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

            log.error("result error    " + result.toString());

            throw new RuntimeException(e);

        }

    }

    private String getRecordTitle(String record) throws Exception {

        String[] recordFields = record.split(",");

        for (String recordField : recordFields) {

            String[] entryPair = recordField.split("\t");

            if (entryPair[0].equals(SOURCE_LABEL) || entryPair[0].equals(TARGET_LABEL)) {

                return entryPair[1];

            }

        }

        return null;

    }

    @Override

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

        super.cleanup(context);

        log.info("Cleaning up mapper...");

    }

    public LodConfiguration getLodConfiguration() {

        return lodConfiguration;

    }

    public void setLodConfiguration(LodConfiguration lodConfiguration) {

        this.lodConfiguration = lodConfiguration;

    }

}