D-Net

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

import eu.dnetlib.data.mapreduce.hbase.lodExport.utils.blocking.Blocking;

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

import eu.dnetlib.data.mapreduce.hbase.lodExport.utils.configuration.Properties;

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;

import java.util.ArrayList;

import java.util.Arrays;

import java.util.Collections;

import java.util.HashSet;

import java.util.List;

import java.util.Set;

/**

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

 * export

 */

/*

-> Parse LOD dump files

        Process lod input files and divide by entity type (both source and target)

        Transform to id, array of [ properties] format

        Store to HDFS

        For -> Multiple outputs and inputs

        Multiple inputs: all source and target datasets and their corresponding mappings

        M/O: separate output files for each dataset: mark records so that they are written to the correct one

*/

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

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

    private LodConfiguration lodConfiguration = new LodConfiguration();

    private String lastExecutionDate;

    private HashSet<String> stopWordsMap = new HashSet<>();

    private boolean createCompositekey;

    private Set<String> usedProperties = new HashSet<>();

    public static enum SOURCE_BUILD_COUNTERS {

        BLOCKING_KEYS,

        REDIS_RECORDS,

        INPUT_SOURCE_RECORDS

    }

    @Override

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

        try {

            lodConfiguration.load(context.getConfiguration().get(Properties.LOD_SOURCE_MAPPINGS));

            stopWordsMap.addAll(Arrays.asList(context.getConfiguration().get(Properties.LOD_STOPWORDS).split(",")));

            createCompositekey = context.getConfiguration().getBoolean(eu.dnetlib.data.mapreduce.hbase.lodExport.utils.configuration.Properties.LOD_CREATE_COMPOSITEKEY, true);

            usedProperties.add("<http://www.eurocris.org/ontologies/cerif/1.3#name>");

            usedProperties.add("<http://lod.openaire.eu/vocab/year>");

        } catch (Exception ex) {

            log.error("An error occured during Mapper Setup " + ex.toString(), ex);

            System.out.println(ex.getCause().toString());

        }

    }

    @Override

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

        try {

            StringBuilder id = new StringBuilder();

            context.getCounter(SOURCE_BUILD_COUNTERS.INPUT_SOURCE_RECORDS).increment(1);

            String[] triples = result.toString().split(Properties.LINE_DELIM);

            List<String> tokenList = parseFieldsAndGenerateTokens(id, triples);

            if (createCompositekey) {

                writeCompositeKeyAndValue(result, context, tokenList, id);

            } else {

                writeSingleKeysAndValues(result, context, tokenList, id);

            }

        } catch (Exception e) {

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

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

            throw new RuntimeException(e);

        }

    }

    private List<String> parseFieldsAndGenerateTokens(StringBuilder id, String[] triples) {

        List<String> tokenList = new ArrayList<>();

        for (String triple : triples) {

            String[] fields = triple.split(Properties.FIELD_DELIM);

            if (fields.length == 3) {

                if (id.length() < 1) {

                    id.append(Properties.SOURCE_PREFIX).append(fields[0]);

                }

                String property = fields[1];

                String value = fields[2];

                if (usedProperties.contains(property)) {

                    String blockingKey = Blocking.tokenBlocking(value, stopWordsMap);

                    tokenList.add(blockingKey);

                }

            }

        }

        return tokenList;

    }

    private void writeSingleKeysAndValues(Text result, Context context, List<String> tokenList, StringBuilder id) throws IOException, InterruptedException {

        for (String token : tokenList) {

            //Write BlockingKey, RecordID to output

            context.write(new Text(token), new Text(id + Properties.FIELD_DELIM + result.toString()));

            context.getCounter(SOURCE_BUILD_COUNTERS.BLOCKING_KEYS).increment(1);

        }

    }

    private void writeCompositeKeyAndValue(Text result, Context context, List<String> tokenList, StringBuilder id) throws IOException, InterruptedException {

        String compositeKey = createCompositeKey(tokenList);

        context.write(new Text(compositeKey), new Text(id + Properties.FIELD_DELIM + result.toString()));

        context.getCounter(SOURCE_BUILD_COUNTERS.BLOCKING_KEYS).increment(1);

    }

    private String createCompositeKey(List<String> tokenList) {

        Collections.sort(tokenList);

        StringBuilder tokenString = new StringBuilder();

        for (String t : tokenList) {

            tokenString.append(t).append(" ");

        }

        return tokenString.toString();

    }

    @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;

    }

    public String getLastExecutionDate() {

        return lastExecutionDate;

    }

    public void setLastExecutionDate(String lastExecutionDate) {

        this.lastExecutionDate = lastExecutionDate;

    }

}