1
\$\begingroup\$

This is an extension of Block Bootstrap Estimation in Java - Part 2. The text file text.txt can be found at https://drive.google.com/open?id=1vLBoNmFyh4alDZt1eoJpavuEwWlPZSKX (please download the file directly should you wish to test it; there are some odd things about it that you will not pick up through even Notepad). This is a small 10 x 10 dataset with maxBlockSize = 10, but this needs to scale up to twenty 5000 x 3000 datasets with maxBlockSize = 3000, just to get an idea of scale.

If helpful, I have 6 cores available with 64GB of RAM, as well as two GPUs that I have no idea how to use (Intel UHD Graphics 630, NVIDIA Quadro P620). I'll be looking over how to use these in the next few days if I have to, and have also considered re-learning Amazon Web Services.

What is clear to me now is that mbbVariance is slowing this program down. Considering how much I need to scale this up, note that mbbVariance and nbbVariance would have to run for each of the 5,000 rows as arguments (x being a row), and have to be iterated (0 to x.length - L + 1 for mbbVariance, and 0 to x.length / L for nbbVariance) proportionally to 5,000 times since maxBlockSize = 5000 implies that values 1, 2, ..., 5000 will be passed as L to the mbbVariance and nbbVariance functions.

While the code below does work, when I ran it for a single 5000 x 3000 data set with maxBlockSize = 3000, I ended up shutting it down after about 10 hours. The nbbThread finished in 18 minutes, judging by the outputted NBB_test.txt. Given how much of MBB_test.txt was populated, through linear interpolation, I estimated that mbbThread would take about 28 hours total (so 18 hours after I had shut the process down). This is too long, given that I have 20 files of this size to process.

import java.io.FileInputStream;
import java.lang.Math;
import java.util.Scanner;
import java.io.IOException;
import java.io.PrintWriter;
import java.io.FileOutputStream;
import java.util.Arrays;
import java.util.stream.IntStream;
import java.util.stream.DoubleStream;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.BrokenBarrierException;
import java.lang.InterruptedException;

public class BlockBootstrapTestParallel {

    // Sum of a subarray, based on B(x, i, L) -- i is one-indexing
    public static double sum(double[] x, int i, int L) {
        return Arrays.stream(x).skip(i - 1).limit(L).sum();
    }

    // Mean of a subarray, based on B(x, i, L) -- i is one-indexing
    public static double mean(double[] x, int i, int L) {
        return Arrays.stream(x).skip(i - 1).limit(L).average().orElse(0);
    }

    // Compute MBB mean
    public static double mbbMu(double[] x, int L) {     
        return IntStream.range(0, x.length - L + 1)
                        .mapToDouble(idx -> mean(x, idx + 1, L))
                        .average()
                        .orElse(0);
    }

    // Compute MBB variance
    public static double mbbVariance(double[] x, int L, double alpha) {
        double mu = mbbMu(x, L);
        double lAlph = Math.pow(L, alpha);

        return IntStream.range(0, x.length - L + 1)
                        .parallel()
                        .mapToDouble(idx -> (lAlph * Math.pow(mean(x, idx + 1, L) - mu, 2)))
                        .average()
                        .orElse(0);
    }

    // Compute NBB mean
    public static double nbbMu(double[] x, int L) {
        return IntStream.range(0, x.length / L)
                        .mapToDouble(idx -> (mean(x, 1 + ((idx + 1) - 1) * L, L)))
                        .average()
                        .orElse(0);
    }

    // Compute NBB variance
    public static double nbbVariance(double[] x, int L, double alpha) {
        double mu = nbbMu(x, L);
        double varSum = IntStream.range(0, x.length / L)
                                 .parallel()
                                 .mapToDouble(idx -> (Math.pow(mean(x, 1 + ((idx + 1) - 1) * L, L) - mu, 2)))
                                 .average()
                                 .orElse(0);

        return Math.pow((double) L, alpha) * varSum;

    }

    // factorial implementation
    public static double factorial(int x) {
        double[] fact = {1.0, 1.0, 2.0, 6.0, 24.0, 120.0, 720.0, 5040.0, 40320.0, 362880.0, 3628800.0};
        return fact[x];
    }

    // Hermite polynomial
    public static double H(double x, int p) {
        double out = IntStream.range(0, (p / 2) + 1)
                              .parallel()
                              .mapToDouble(idx -> (Math.pow(-1, idx) * Math.pow(x, p - (2 * idx)) / 
                                                    ((factorial(idx) * factorial(p - (2 * idx))) * (1L << idx))))
                              .sum();
        out *= factorial(p);
        return out;
    }

    // Row means
    public static double[] rowMeans(double[][] x, int nrows, int ncols) {
        return IntStream.range(0, nrows)
                        .parallel()
                        .mapToDouble(idx -> (mean(x[idx], 1, ncols)))
                        .toArray();
    }

    public static void duration(long start, long end) {
        System.out.println("Total execution time: " + (((double)(end - start))/60000) + " minutes");
    }


    public static void main(String[] argv) throws InterruptedException, BrokenBarrierException, IOException {
        final long start = System.currentTimeMillis();
        FileInputStream fileIn = new FileInputStream("test.txt");
        FileOutputStream fileOutMBB = new FileOutputStream("MBB_test.txt");
        FileOutputStream fileOutNBB = new FileOutputStream("NBB_test.txt");
        FileOutputStream fileOutMean = new FileOutputStream("means_test.txt");

        Scanner scnr = new Scanner(fileIn);
        PrintWriter outFSMBB = new PrintWriter(fileOutMBB);
        PrintWriter outFSNBB = new PrintWriter(fileOutNBB);
        PrintWriter outFSmean = new PrintWriter(fileOutMean);

        // These variables are taken from the command line, but are inputted here for ease of use.
        int rows = 10;
        int cols = 10;
        int maxBlockSize = 10; // this could potentially be any value <= cols
        int p = 1;
        double alpha = 0.1;
        double[][] timeSeries = new double[rows][cols];

        // read in the file, and perform the H_p(x) transformation
        for (int i = 0; i < rows; i++) {
            for (int j = 0; j < cols; j++) {
                timeSeries[i][j] = H(scnr.nextDouble(), p);
            }
            scnr.next(); // skip null terminator
        }

        // row means
        double[] sampleMeans = rowMeans(timeSeries, rows, cols);
        for (int i = 0; i < rows; i++) {
            outFSmean.print(sampleMeans[i] + " ");
        }
        outFSmean.println();
        outFSmean.close();

        final CyclicBarrier gate = new CyclicBarrier(3);

        Thread mbbThread = new Thread(() -> {
            try {
                gate.await();
                for (int j = 0; j < rows; j++) {
                    for (int m = 0; m < maxBlockSize; m++) {
                        outFSMBB.print(mbbVariance(timeSeries[j], m + 1, alpha) + " ");
                    }
                outFSMBB.println();
                }           
                outFSMBB.close();
            } catch (InterruptedException e) {
                System.out.println("Main Thread interrupted!");
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                System.out.println("Main Thread interrupted!");
                e.printStackTrace();
            }
        });

        Thread nbbThread = new Thread(() -> {
            try {
                gate.await();
                for (int j = 0; j < rows; j++) {
                    for (int m = 0; m < maxBlockSize; m++) {
                        outFSNBB.print(nbbVariance(timeSeries[j], m + 1, alpha) + " ");
                    }
                outFSNBB.println();
                }           
                outFSNBB.close();
            } catch (InterruptedException e) {
                System.out.println("Main Thread interrupted!");
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                System.out.println("Main Thread interrupted!");
                e.printStackTrace();
            }
        });

        // start threads simultaneously
        mbbThread.start();
        nbbThread.start();
        gate.await();

        // wait for threads to die 
        mbbThread.join();
        nbbThread.join();

        duration(start, System.currentTimeMillis());
    }
}
\$\endgroup\$

0

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Browse other questions tagged or ask your own question.