D-Net

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

import eu.dnetlib.data.mapreduce.hbase.lodExport.utils.caching.RedisUtils;

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

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

import org.apache.log4j.Logger;

        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

*/

    private Map<String, Integer> stopWordsMap = new HashMap<>();

    private static final String FIELD_DELIM = "\t";

    @Override

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

            lodConfiguration = new LodConfiguration();

            lodConfiguration.load(context.getConfiguration().get("lod.targetMappings"));

            stopWords = context.getConfiguration().get("lod.stopwords");

            for (String stopword : stopWords.split(",")) {

                stopWordsMap.put(stopword, 0);

            }

        } catch (Exception ex) {

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

        try {

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

                    if (id.length() < 1) {

                        id.append("target_").append(fields[0]);

                    }

                        List<String> blockingKeys = Blocking.tokenBlocking(value, stopWordsMap);

                        //Write BlockingKey, RecordID to output

                        for (String blockingKey : blockingKeys) {

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

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

                        }

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

            throw new RuntimeException(e);

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

        super.cleanup(context);

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

    }

        return log;

    }

        this.log = log;

    }

    public LodConfiguration getLodConfiguration() {

        return lodConfiguration;

    }

    public void setLodConfiguration(LodConfiguration lodConfiguration) {

        this.lodConfiguration = lodConfiguration;

    }