Create GaskySparkReducerJob

spark/src/main/java/com/css534/parallel/GaskySparkReducerJob

@@ -0,0 +1,387 @@
+import org.apache.spark.api.java.JavaPairRDD;
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.JavaSparkContext;
+import scala.Tuple2;
+import org.apache.spark.SparkConf;
+import org.apache.spark.api.java.function.Function;
+import java.util.Collections;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.List;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+
+public class GaskySparkJob {
+    private static final int GRID_SIZE = 8; 
+    public static void main(String[] args) {
+        // Create a Spark context
+        SparkConf conf = new SparkConf().setAppName("GaskySparkJob");
+        JavaSparkContext sparkContext = new JavaSparkContext(conf);
+
+        // Load input data
+        JavaRDD<String> inputData = sparkContext.textFile(args[0]);
+
+        // Debug: Print the input data
+        System.out.println("Debug: Input Data:");
+        inputData.foreach(line -> System.out.println(line));
+
+        // Process the input data and create key-value pairs
+        JavaPairRDD<Tuple2<String, Integer>, Iterable<Tuple2<Integer, Double>>> result = inputData
+                .flatMapToPair(line -> parseInputData(line))
+                .groupByKey();
+
+        // Collect the result and sort it
+        List<Tuple2<Tuple2<String, Integer>, Iterable<Tuple2<Integer, Double>>>> resultList = result.collect();
+
+        // Create a new list with the sorted elements
+        List<Tuple2<Tuple2<String, Integer>, Iterable<Tuple2<Integer, Double>>>> sortedResultList = new ArrayList<>(resultList);
+        sortedResultList.sort(Comparator.comparing(
+                (Tuple2<Tuple2<String, Integer>, Iterable<Tuple2<Integer, Double>>> tuple) -> tuple._1(),
+                Tuple2Comparator.INSTANCE
+        ));
+
+        // Separate keys and values from the sorted result
+        List<Tuple2<String, Integer>> keysList = new ArrayList<>();
+        List<Iterable<Tuple2<Integer, Double>>> valuesList = new ArrayList<>();
+
+        for (Tuple2<Tuple2<String, Integer>, Iterable<Tuple2<Integer, Double>>> tuple : sortedResultList) {
+            keysList.add(tuple._1());
+            valuesList.add(tuple._2());
+        }
+
+        // Example: Print keys and values
+        System.out.println("Keys:");
+        keysList.forEach(key -> System.out.println(key));
+
+        System.out.println("Values:");
+        valuesList.forEach(values -> {
+            values.forEach(value -> System.out.print("  " + value));
+            System.out.println("");
+        });
+
+        // Call the new method to apply gaskyAlgorithm
+        applyGaskyAlgorithm(sparkContext,keysList, valuesList);
+
+        // Stop the Spark context
+        sparkContext.stop();
+    }
+
+        private static void applyGaskyAlgorithm(JavaSparkContext sparkContext, List<Tuple2<String, Integer>> keysList, List<Iterable<Tuple2<Integer, Double>>> valuesList) {
+        System.out.println("valuesList size" + valuesList.size());
+
+        // Create JavaRDDs for keys and values
+        JavaRDD<Tuple2<String, Integer>> keysRDD = sparkContext.parallelize(keysList);
+        JavaRDD<Iterable<Tuple2<Integer, Double>>> valuesRDD = sparkContext.parallelize(valuesList);
+
+        // Zip the two RDDs to create a PairRDD
+        JavaPairRDD<Tuple2<String, Integer>, Iterable<Tuple2<Integer, Double>>> zippedRDD = keysRDD.zip(valuesRDD);
+
+        // Filter out points with Double.MAX_VALUE distance
+        JavaPairRDD<Tuple2<String, Integer>, Iterable<Tuple2<Integer, Double>>> filteredRDD = zippedRDD.mapValues(values ->
+                filterUnfavorablePoints(values)
+        );
+
+        // Debug: Print the filtered points
+        System.out.println("Remaining Points after filtering:");
+        filteredRDD.collect().forEach(tuple -> {
+        Tuple2<String, Integer> keysTuple = tuple._1();
+        int colNumber = keysTuple._2();
+        Iterable<Tuple2<Integer, Double>> remainingPoints = tuple._2();
+        System.out.println("Column number: " + colNumber);
+        System.out.println("Remaining Points: " + remainingPoints);
+        }); 
+
+        // Apply gaskyAlgorithm to the remaining points
+        filteredRDD.foreach(tuple -> {
+            Tuple2<String, Integer> keysTuple = tuple._1();
+            int colNumber = keysTuple._2();
+            Iterable<Tuple2<Integer, Double>> remainingPoints = tuple._2();
+
+            System.out.println("Applying gaskyAlgorithm for Column number: " + colNumber);
+            // Assuming gaskyAlgorithm takes an Iterable<Tuple2<Integer, Double>> and a Tuple2<String, Integer>
+            gaskyAlgorithm(remainingPoints, colNumber);
+        });
+    }
+
+    private static Iterable<Tuple2<Integer, Double>> filterUnfavorablePoints(Iterable<Tuple2<Integer, Double>> values) {
+        List<Tuple2<Integer, Double>> filteredValues = new ArrayList<>();
+        for (Tuple2<Integer, Double> value : values) {
+            if (!value._2().equals(Double.MAX_VALUE)) {
+                filteredValues.add(value);
+            }
+        }
+        return filteredValues;
+    }
+
+
+
+    private static void gaskyAlgorithm(Iterable<Tuple2<Integer, Double>> remainingPoints, int colNumber) {
+    int totalRemainingPoints = Iterables.size(remainingPoints);
+    List<Tuple2<Double, Double>> points = new ArrayList<>();
+
+    // Convert remainingPoints to Tuple2<Double, Double> objects
+    for (Tuple2<Integer, Double> point : remainingPoints) {
+        points.add(new Tuple2<>(point._1().doubleValue(), point._2()));
+    }
+
+    // Debug: Print initial points
+    System.out.println("Initial Points for Column number: " + colNumber);
+    points.forEach(System.out::println);
+
+    // Filtering based on dominance
+    int currentWindowStart = 1;
+    while (points.size() >= 3 && currentWindowStart <= points.size() - 2) {
+        Tuple2<Double, Double> ii = points.get(currentWindowStart - 1);
+        Tuple2<Double, Double> jj = points.get(currentWindowStart);
+        Tuple2<Double, Double> kk = points.get(currentWindowStart + 1);
+
+        if (ii != null && jj != null && kk != null) {
+            double xij = calcBisectorProjections(ii._1(), ii._2(), jj._1(), jj._2())._1();
+            double xjk = calcBisectorProjections(jj._1(), jj._2(), kk._1(), kk._2())._1();
+
+            // Debug: Print xij and xjk
+            System.out.println("xij: " + xij + ", xjk: " + xjk);
+
+            if (xij > xjk) {
+                // Debug: Print the removed point
+                System.out.println("Removed point: " + jj);
+
+                // Remove the middle point
+                points.remove(currentWindowStart);
+            } else {
+                // Move to the next window
+                currentWindowStart++;
+            }
+        }
+    }
+
+    // Debug: Print final points after filtering
+    System.out.println("Filtered Points after Gasky Algorithm for Column number: " + colNumber);
+    points.forEach(System.out::println);
+
+    // Call the method to find proximal points
+    List<double[]> proximityProjectionsPoints = findProximityPoints(points, totalRemainingPoints, GRID_SIZE);
+
+    // Debug: Print proximal points
+    System.out.println("Proximal Points for Column number: " + colNumber);
+    proximityProjectionsPoints.forEach(point -> System.out.println(Arrays.toString(point)));
+
+    int unDominatedPointsSize = points.size();
+    int proximityIntervals = proximityProjectionsPoints.size() - 1;
+    int dominatedCoordinatesDistances = 0;
+    List<Double> distances = new ArrayList<>(Collections.nCopies(GRID_SIZE, Double.MAX_VALUE));
+    //update distance implementation
+    for (int interval = 0; interval < proximityIntervals; interval++) {
+            double[] currentInterval = proximityProjectionsPoints.get(interval);
+            Tuple2<Double,Double> dominantPoint = points.get(dominatedCoordinatesDistances);
+            int start = (int) currentInterval[0];
+            int end = (int) currentInterval[1];
+            for (int xCord = start; xCord <= end; xCord++) {
+            //     if (xCord >= 1 && xCord <= distances.size()) {
+            //         double currentDistance = distances.get(xCord - 1);
+            //         double newDistance = findEuclideanDistance(xCord, 0, currentInterval[2], currentInterval[3]);
+
+            //         if (currentDistance != Double.MAX_VALUE) {
+            //             newDistance = Math.min(newDistance, currentDistance);
+            //         }
+
+            //         distances.set(xCord - 1, newDistance);
+            //     }
+            // }
+
+            // dominatedCoordinatesDistances++;
+                    if (distances.get(xCord - 1) != Double.MAX_VALUE){
+                        distances.set(
+                                xCord - 1,
+                                Double.min(
+                                        findEuclideanDistance(xCord, 0, dominantPoint._1(), dominantPoint._2()),
+                                        distances.get(xCord - 1)
+                                )
+                        );
+                    }else
+                        distances.set(
+                                xCord - 1,
+                                findEuclideanDistance(xCord, 0, dominantPoint._1(), dominantPoint._2())
+                        );
+                }
+                dominatedCoordinatesDistances++;
+        }
+
+     // Debug: Print updated distances
+        System.out.println("Updated Distances:");
+        distances.forEach(System.out::println);
+
+
+}
+
+    private static List<double[]> findProximityPoints(List<Tuple2<Double, Double>> unDominatedPoints, final int totalPoints, final int gridSize) {
+    List<Tuple2<Double, Double>> intervals = new ArrayList<>();
+
+    System.out.println("unDominatedPoints size: " + unDominatedPoints.size());
+    // Calculate intervals based on the unDominatedPoints
+    for (int i = 1; i < unDominatedPoints.size(); i++) {
+        Tuple2<Double, Double> point1 = unDominatedPoints.get(i - 1);
+        Tuple2<Double, Double> point2 = unDominatedPoints.get(i);
+        intervals.add(new Tuple2<>((point1._1() + point2._1()) / 2, 0.0));
+    }
+
+    // Combine intervals using a frame
+    List<double[]> mergedInterval = new ArrayList<>(intervals.size());
+    mergedInterval.add(new double[]{1, intervals.get(0)._1()});
+
+    for (int i = 1; i < intervals.size(); i++) {
+        mergedInterval.add(new double[]{intervals.get(i - 1)._1(), intervals.get(i)._1()});
+    }
+
+    mergedInterval.add(new double[]{intervals.get(intervals.size() - 1)._1(), GRID_SIZE});
+
+    return mergedInterval;
+}
+
+    private static double findEuclideanDistance(int x, int y, int x1, int y1) {
+        return Math.sqrt((x1 - x) * (x1 - x) + (y1 - y) * (y1 - y));
+    }
+
+    private static double findEuclideanDistance(double x, double y, double x1, double y1){
+        return Math.sqrt((x1 - x) * (x1 - x) + (y1 - y) * (y1 - y));
+    }
+
+    // FlatMap function to process input data and generate key-value pairs
+    private static Iterator<Tuple2<Tuple2<String, Integer>, Tuple2<Integer, Double>>> parseInputData(String line) {
+        System.out.println("Debug: Processing Line: " + line);
+
+        String[] distFavArray = line.split("\\s+");
+        List<Tuple2<Tuple2<String, Integer>, Tuple2<Integer, Double>>> result = new ArrayList<>();
+
+        if (distFavArray.length > 0) {
+            String facilityName = distFavArray[0];
+            int matrixRowNumber = Integer.parseInt(distFavArray[1]);
+
+            // Convert the strings to a list of strings
+            List<String> binMatrixValues = Arrays.asList(Arrays.copyOfRange(distFavArray, 2, distFavArray.length));
+
+            double[] leftDistance = new double[binMatrixValues.get(0).length()];
+            double[] rightDistance = new double[binMatrixValues.get(0).length()];
+
+            Arrays.fill(leftDistance, Double.MAX_VALUE);
+            Arrays.fill(rightDistance, Double.MAX_VALUE);
+
+            leftDistance = getLeftDistance(leftDistance, binMatrixValues);
+            rightDistance = getRightDistance(rightDistance, binMatrixValues);
+
+            for (int i = 0; i < binMatrixValues.get(0).length(); i++) {
+                result.add(new Tuple2<>(new Tuple2<>(facilityName, i + 1),
+                        new Tuple2<>(matrixRowNumber, Double.min(leftDistance[i], rightDistance[i]))));
+            }
+        }
+
+        // Debug: Print the generated key-value pairs
+        System.out.println("Debug: Generated Key-Value Pairs:");
+        result.forEach(tuple -> System.out.println(tuple));
+
+        return result.iterator();
+    }
+
+    private static double[] getLeftDistance(double[] leftDistance, List<String> gridRows) {
+        boolean isFavlFound = false;
+        for (int i = 0; i < gridRows.get(0).length(); i++) {
+            if (gridRows.get(0).charAt(i) == '1') {
+                leftDistance[i] = 0;
+                isFavlFound = true;
+            } else if (isFavlFound) {
+                leftDistance[i] = leftDistance[i - 1] + 1;
+            }
+        }
+        return leftDistance;
+    }
+
+    private static double[] getRightDistance(double[] rightDistance, List<String> gridRows) {
+        boolean isFavrFound = false;
+        for (int i = gridRows.get(0).length() - 1; i >= 0; --i) {
+            if (gridRows.get(0).charAt(i) == '1') {
+                rightDistance[i] = 0;
+                isFavrFound = true;
+            } else if (isFavrFound) {
+                rightDistance[i] = rightDistance[i + 1] + 1;
+            }
+        }
+        return rightDistance;
+    }
+
+    // Comparator for Tuple2<String, Integer>
+    static class Tuple2Comparator implements Comparator<Tuple2<String, Integer>>, Serializable {
+        static final Tuple2Comparator INSTANCE = new Tuple2Comparator();
+
+        @Override
+        public int compare(Tuple2<String, Integer> tuple1, Tuple2<String, Integer> tuple2) {
+            int compareResult = tuple1._1().compareTo(tuple2._1());
+            if (compareResult == 0) {
+                // If the first elements are equal, compare the second elements
+                compareResult = Integer.compare(tuple1._2(), tuple2._2());
+            }
+            return compareResult;
+        }
+    }
+
+
+
+    private static Tuple2<Double, Double> calcBisectorProjections(double x, double y, double x1, double y1) {
+        double xx = ((y1 * y1) - (y * y) + (x1 * x1) - (x * x)) / (2 * (x1 - x));
+        double yy = 0;
+        return new Tuple2<>(xx, yy);
+    }
+
+//     private static List<double[]> findProximityPoints(JavaSparkContext sparkContext,
+//                                                   List<Tuple2<Double, Double>> unDominatedPoints,
+//                                                   final int totalPoints) {
+//     // Create a JavaRDD from the list of unDominatedPoints
+//     JavaRDD<Tuple2<Double, Double>> unDominatedPointsRDD = sparkContext.parallelize(unDominatedPoints);
+
+//     // Calculate intervals between consecutive points
+//     JavaRDD<Tuple2<Double, Double>> intervalsRDD = unDominatedPointsRDD
+//             .zipWithIndex()
+//             .filter(tuple -> tuple._2() > 0) // Exclude the first point
+//             .map(tuple -> {
+//                 Tuple2<Double, Double> point1 = unDominatedPoints.get(tuple._2().intValue() - 1);
+//                 Tuple2<Double, Double> point2 = tuple._1();
+
+//                 return new Tuple2<>(
+//                         (point1._1() + point2._1()) / 2,
+//                         0.0 // point lying on with intersection on X axis
+//                 );
+//             });
+
+//     // Combine intervals
+//     JavaPairRDD<Double, Double> mergedIntervalRDD = intervalsRDD
+//             .mapToPair(interval -> new Tuple2<>(interval._1(), interval._1()))
+//             .reduceByKey((interval1, interval2) -> new Tuple2<>(interval1, interval2));
+
+//     // Collect the results back to the driver
+//     List<Tuple2<Double, Double>> mergedIntervals = mergedIntervalRDD.collect();
+
+//     // Convert List<Tuple2<Double, Double>> to List<double[]>
+//     List<double[]> mergedIntervalList = new ArrayList<>();
+//     for (Tuple2<Double, Double> interval : mergedIntervals) {
+//         mergedIntervalList.add(new double[]{interval._1(), totalPoints});
+//     }
+
+//     return mergedIntervalList;
+// }
+
+// Add this method to your class
+private static <T> Iterable<T> iterableToJava(scala.collection.Iterable<T> scalaIterable) {
+    List<T> javaList = new ArrayList<>();
+    scala.collection.Iterator<T> scalaIterator = scalaIterable.iterator();
+    while (scalaIterator.hasNext()) {
+        javaList.add(scalaIterator.next());
+    }
+    return javaList;
+}
+
+}
+