D-Net

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

import com.lambdaworks.redis.RedisClient;

import com.lambdaworks.redis.RedisConnection;

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

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;

import java.util.ArrayList;

import java.util.Iterator;

import java.util.List;

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

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

    private MemoryCache sourceCache, targetCache;

    private KBInfo sourceKb, targetKb;

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

    private String redisHost;

    private Integer redisPort;

    private RedisConnection connection;

    private RedisClient client;

    private long optimalBlockSize;

    public static enum LIMES_COUNTERS {

        TARGET_TRIPLES,

        SOURCE_TRIPLES,

        WRITTEN_OUT_ENTITIES,

        LINKED_BLOCKS,

        TOTAL_BLOCKS,

        S_IN,

        T_IN,

        DISCARDED_BLOCKS

    }

    @Override

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

        try {

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

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

            XMLConfigurationReader reader = new XMLConfigurationReader();

            config = reader.read(configXML, limesDTD);

            sourceCache = new MemoryCache();

            targetCache = new MemoryCache();

            sourceKb = config.getSourceInfo();

            targetKb = config.getTargetInfo();

            redisHost = context.getConfiguration().get("lod.redisHost");

            redisPort = Integer.parseInt(context.getConfiguration().get("lod.redisPort"));

            client = RedisClient.create("redis://" + redisHost + ":" + redisPort);

            connection = client.connect();

            String statsPath = context.getConfiguration().get("lod.statsOutputPath");

            FrequencyCounter frequencyCounter = new FrequencyCounter();

            this.optimalBlockSize = frequencyCounter.getOptimalBlockSize(statsPath);

            //this.optimalBlockSize = 17000;

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

        } 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 {

        Iterator<Text> recordsIterator = values.iterator();

//each list ia a block now

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

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

        int totalRecords = 0;

        boolean hasSource = false;

        boolean hasTarget = false;

        while (recordsIterator.hasNext()) {

            String recordId = recordsIterator.next().toString();

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

                hasSource = true;

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

            }

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

                hasTarget = true;

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

            }

            records.add(recordId);

            totalRecords++;

        }

        if (totalRecords > 1 && totalRecords < optimalBlockSize && hasSource && hasTarget) {

            try {

                fillLimesCache(records, context);

                linkRecords(context);

            } catch (Exception e) {

                log.error(e.toString());

                throw new IOException(e);

            }

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

        } else {

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

        }

    }

    private void fillLimesCache(List<String> records, Context context) {

        for (String record : records) {

            String[] Fields = ((String) connection.get(record)).split(",");

            String subject = record.substring(record.indexOf("_") + 1);

            String[] split;

            for (int j = 0; j < Fields.length; j++) {

                split = Fields[j].split("\t");

                String field = split[0].replace("\t", "").trim();

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

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

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

                            String processedValue = Preprocessor.process(split[1], sourceKb.getFunctions().get(field).get(propertyDub));

                            sourceCache.addTriple(subject, propertyDub, processedValue);

                        }

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

                    } else {

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

                            String processedValue = Preprocessor.process(split[1], targetKb.getFunctions().get(field).get(propertyDub));

                            targetCache.addTriple(subject, propertyDub, processedValue);

                        }

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

                    }

                }

            }

        }

        records.clear();

        }

    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.replace("source_", "")), new Text(target.replace("target_", "") + "," + acceptanceMapping.getConfidence(source, target)));

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

            }

        }

    }

    @Override

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

    }

}