D-Net

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

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

import org.aksw.limes.core.execution.engine.ExecutionEngine;

import org.aksw.limes.core.execution.engine.ExecutionEngineFactory;

import org.aksw.limes.core.execution.planning.plan.NestedPlan;

import org.aksw.limes.core.execution.planning.planner.ExecutionPlannerFactory;

import org.aksw.limes.core.execution.planning.planner.IPlanner;

import org.aksw.limes.core.execution.rewriter.Rewriter;

import org.aksw.limes.core.execution.rewriter.RewriterFactory;

import org.aksw.limes.core.io.cache.MemoryCache;

import org.aksw.limes.core.io.config.KBInfo;

import org.aksw.limes.core.io.config.reader.xml.XMLConfigurationReader;

import org.aksw.limes.core.io.ls.LinkSpecification;

import org.aksw.limes.core.io.mapping.Mapping;

import org.aksw.limes.core.io.preprocessing.Preprocessor;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

import org.apache.log4j.Logger;

import java.io.IOException;

public class LimesReducer extends Reducer<Text, Text, Text, Text> {

    private Logger log = Logger.getLogger(LimesReducer.class);

    private MemoryCache sourceCache, targetCache;

    private KBInfo sourceKb, targetKb;

    private org.aksw.limes.core.io.config.Configuration config;

    private RedisUtils redisUtils;

    private int optimalBlockSize;

    private static final String OPTIMAL_BLOCKS = "optimalBlockSize";

    private static final String SEPERATOR = ",";

    private static final String LINE_SEPERATOR = "\t.\t";

    private static final String FIELD_DELIM = "\t";

    public static enum LIMES_COUNTERS {

        TARGET_TRIPLES,

        SOURCE_TRIPLES,

        WRITTEN_OUT_ENTITIES,

        LINKED_BLOCKS,

        DISCARDED_BLOCKS

    }

    @Override

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

        try {

            redisUtils = new RedisUtils(context);

            String configXML = context.getConfiguration().get("lod.configXML");

            String limesDTD = context.getConfiguration().get("lod.limesDTD");

            XMLConfigurationReader reader = new XMLConfigurationReader();

            config = reader.read(configXML, limesDTD);

            sourceKb = config.getSourceInfo();

            targetKb = config.getTargetInfo();

            optimalBlockSize = Integer.valueOf(redisUtils.getValue(OPTIMAL_BLOCKS));

            log.info("OPTIMAL BLOCK SIZE " + optimalBlockSize);

            //    optimalBlockSize = 1700;

            log.info("sourceKB" + sourceKb.getProperties());

            log.info("targetKB" + targetKb.getProperties());

        } catch (Exception e) {

            log.error("Error computing stats" + e.toString());

            throw new RuntimeException(e);

        }

    }

    @Override

    protected void reduce(final Text key, final Iterable<Text> values, final Context context) throws IOException, InterruptedException {

        //each item in the list is a block with the given key

        for (Text block : values) {

            try {

                context.getCounter(LIMES_COUNTERS.LINKED_BLOCKS).increment(1);

                sourceCache = new MemoryCache();

                targetCache = new MemoryCache();

                fillLimesCache(block, context);

                linkRecords(context);

            } catch (Exception e) {

                log.error(e.toString(), e);

                throw new IOException(e);

            }

        }

    }

    private void fillLimesCache(Text block, Context context) throws Exception {

        String[] split = block.toString().split(SEPERATOR);

        for (String recordId : split) {

            String record = redisUtils.getValue(recordId);

            if (record == null) {

                log.error("Record " + recordId + " not found! ");

                throw new Exception("Record " + recordId + " not found! ");

            }

            record = record.substring(record.indexOf(FIELD_DELIM) + 1).trim();

            String[] Triples = record.split(LINE_SEPERATOR);

            for (String triple : Triples) {

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

                String subject = Fields[0];

                //props in sourceKB do not contain <> ,so we need to replace them.

                String property = Fields[1].replaceAll("[<>]", "");

                String value = Fields[2];

                if (sourceKb.getProperties().contains(property) || targetKb.getProperties().contains(property)) {

                    if (recordId.contains("source_")) {

                        for (String propertyDub : sourceKb.getFunctions().get(property).keySet()) {

                            String processedValue = Preprocessor.process(value, sourceKb.getFunctions().get(property).get(propertyDub));

                            sourceCache.addTriple(subject, propertyDub, processedValue);

                        }

                        context.getCounter(LIMES_COUNTERS.SOURCE_TRIPLES).increment(1);

                    } else {

                        for (String propertyDub : targetKb.getFunctions().get(property).keySet()) {

                            String processedValue = Preprocessor.process(value, targetKb.getFunctions().get(property).get(propertyDub));

                            targetCache.addTriple(subject, propertyDub, processedValue);

                        }

                        context.getCounter(LIMES_COUNTERS.TARGET_TRIPLES).increment(1);

                    }

                }

            }

        }

    }

    private void linkRecords(Context context) throws IOException, InterruptedException {

        //link specification

        Rewriter rw = RewriterFactory.getRewriter("Default");

        LinkSpecification ls = new LinkSpecification(config.getMetricExpression(), config.getVerificationThreshold());

        LinkSpecification rwLs = rw.rewrite(ls);

        //planning

        IPlanner planner = ExecutionPlannerFactory.getPlanner(config.getExecutionPlan(), sourceCache, targetCache);

        assert planner != null;

        NestedPlan plan = planner.plan(rwLs);

        ExecutionEngine engine = ExecutionEngineFactory.getEngine("Default", sourceCache, targetCache,

                config.getSourceInfo().getVar(), config.getTargetInfo().getVar());

        Mapping verificationMapping = engine.execute(plan); //mappings for verification

        Mapping acceptanceMapping = verificationMapping.getSubMap(config.getAcceptanceThreshold()); //mappings for acceptance (auta theloume)

        //output

        for (String source : acceptanceMapping.getMap().keySet()) {//gia kathe source blepoume ta targets

            for (String target : acceptanceMapping.getMap().get(source).keySet()) {//gia kathe target blepoume to confidence

                context.write(new Text(source), new Text(target + "," + acceptanceMapping.getConfidence(source, target)));

                context.getCounter(LIMES_COUNTERS.WRITTEN_OUT_ENTITIES).increment(1);

            }

        }

    }

    @Override

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

    }

}