D-Net

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

import com.google.common.base.Functions;

import com.google.common.collect.ImmutableSortedMap;

import com.google.common.collect.Ordering;

import com.lambdaworks.redis.RedisClient;

import com.lambdaworks.redis.RedisConnection;

import com.lambdaworks.redis.RedisStringsConnection;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.LocatedFileStatus;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.fs.RemoteIterator;

import org.apache.hadoop.io.SequenceFile;

import org.eclipse.jdt.internal.core.SourceType;

import org.junit.Test;

import sun.reflect.generics.tree.Tree;

import java.io.*;

import java.math.BigInteger;

import java.util.*;

import java.io.BufferedReader;

import java.io.BufferedWriter;

import java.io.IOException;

import java.io.InputStreamReader;

import java.io.OutputStreamWriter;

import java.util.ArrayList;

import java.util.HashMap;

import java.util.List;

import java.util.Map;

public class FrequencyCounter {

    private enum ORDERING_TYPE {

        ASC, DESC

    }

    public static void main(String[] args) throws Exception {

        String inputPath = "/tmp/lod_blocks/stats";

        String outputPath = "/tmp/lod_blocks/stats/frequencyMap";

        String redisHost = "194.177.192.118";

        try {

            FileSystem fs = FileSystem.get(new Configuration());

            Path path = new Path(inputPath);

            List filenames = getFiles(fs, path);

            TreeMap<Integer, Integer> frequencyMap = getFrequencyMap(fs, filenames);

            System.out.println("Sorted Map is " + frequencyMap.entrySet());

            TreeMap<BigInteger, Double> statistics = getStatistics(frequencyMap);

            int optimalBlockSize = getOptimalBlockSize(statistics);

            //WRITE optimalBlockSize to REDIS? somewhere .. Goes here!!!

            System.out.println("Statistics Map" + statistics.entrySet());

            System.out.println("Optimal Size " + optimalBlockSize);

            writeToRedis("optimalBlockSize", String.valueOf(optimalBlockSize));

            fs.close();

        } catch (Exception e) {

            throw new Exception("Error : ", e);

        }

    }

    private static List getFiles(FileSystem fs, Path path) throws IOException {

        RemoteIterator<LocatedFileStatus> Files;

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

        Files = fs.listFiles(path, false);

        while (Files.hasNext()) {

            String fileName = Files.next().getPath().toString();

            fileNames.add(fileName);

        }

        System.out.println("Filenames" + fileNames);

        return fileNames;

    }

    private static TreeMap<Integer, Integer> getFrequencyMap(FileSystem fs, List<String> fileNames) throws Exception {

        String line = null;

        try {

            TreeMap<Integer, Integer> frequencyMap = new TreeMap<>();

            for (String file : fileNames) {

                System.out.println("opening " + new Path(file));

                BufferedReader br = new BufferedReader(new InputStreamReader(fs.open(new Path(file))));

                line = br.readLine();

                while (line != null) {

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

                    if (frequencyMap.containsKey(Integer.parseInt(split[1]))) {

                        frequencyMap.put(Integer.parseInt(split[1]), frequencyMap.get(Integer.parseInt(split[1])) + 1);

                    } else {

                        frequencyMap.put(Integer.parseInt(split[1]), 1);

                    }

                    line = br.readLine();

                }

            }

            return frequencyMap;

        } catch (Exception e) {

            throw new Exception("Error : " + line, e);

        }

    }

    private static TreeMap<BigInteger, Double> getStatistics(TreeMap<Integer, Integer> sortedMap) {

        TreeMap<BigInteger, Double> statistics = new TreeMap<BigInteger, Double>();

        double CC = 0d;

        CC = 0d;

        int f;

        BigInteger totalSizeOfBlocks = BigInteger.ZERO;

        BigInteger numberOfComparisons = BigInteger.ZERO;

        Set<Integer> keys = sortedMap.keySet();

        for (Integer key : keys) {

            BigInteger blockSize = BigInteger.valueOf(key.intValue());

            f = sortedMap.get(key);

            totalSizeOfBlocks = totalSizeOfBlocks.add(BigInteger.valueOf(f).multiply(blockSize));

            numberOfComparisons = numberOfComparisons.add(BigInteger.valueOf(f).multiply(blockSize.multiply(blockSize.subtract(BigInteger.ONE)).shiftLeft(1)));

            CC = totalSizeOfBlocks.doubleValue() / numberOfComparisons.doubleValue();

            statistics.put(blockSize, CC);

        }

        return statistics;

    }

    private static int getOptimalBlockSize(TreeMap<BigInteger, Double> statistics) {

        int optimalBlockSize = Integer.valueOf(statistics.lastEntry().getKey().intValue());

        double eps = 1d;

        NavigableSet<BigInteger> keys = statistics.descendingKeySet();

        for (BigInteger key : keys) {

        	double diff = Math.abs(statistics.get(key) - statistics.get(key.subtract(BigInteger.ONE)));

            if (diff < eps) {

                eps = diff;

                optimalBlockSize = Integer.valueOf(key.intValue());

            }

        }

        return optimalBlockSize;

    }

    public static void writeToRedis(String key, String value) {

        RedisClient client = RedisClient.create("redis://194.177.192.118");

        RedisStringsConnection<String, String> connection = client.connect();

        connection.set(key, value);

        System.out.println("Set to redis " + " " + key + " to " + connection.get(key));

    }

    public static SortedMap<Integer, Integer> sortMap(Map map, ORDERING_TYPE ordering) {

        Ordering<Integer> valueComparator;

        if (ordering == ORDERING_TYPE.DESC) {

            valueComparator = Ordering.natural().onResultOf(Functions.forMap(map)).compound(Ordering.natural()).reverse();

        } else {

            valueComparator = Ordering.natural().onResultOf(Functions.forMap(map)).compound(Ordering.natural());

        }

        return ImmutableSortedMap.copyOf(map, valueComparator);

    }

    private static void writeMap(FileSystem fs, TreeMap<Integer, Integer> map, String output) throws IOException {

        Path outputPath = new Path(output);

        BufferedWriter br = new BufferedWriter(new OutputStreamWriter(fs.create(outputPath, true)));

        // TO append data to a file, use fs.append(Path f)

        for (Map.Entry<Integer, Integer> entry : map.entrySet()) {

            br.append(entry.getKey() + ":" + entry.getValue() + ",");

        }

        br.close();

    }

/*   @Test

    public void test() {

        TreeMap map = new TreeMap();

        map.put(6, 23);

        map.put(3, 9);

        map.put(1, 2);

        System.out.println(map.entrySet());

       System.out.println(map.descendingMap());

    }*/

}