Actitacker Spark Cassandra

PredictActivity.java

private static Vector computeFeature(JavaSparkContext sc) {

        double[] features = new double[11];

        // retrieve data from Cassandra and create an CassandraRDD
        CassandraJavaRDD<CassandraRow> cassandraRowsRDD = javaFunctions(sc).cassandraTable("accelerations", "acceleration");

        JavaRDD<CassandraRow> data = cassandraRowsRDD.select("timestamp", "x", "y", "z")
                .where("user_id=?", "MY_USER") //user ID is hard coded in REST API app
                .withDescOrder()
                .limit(500l); //load the last 500 acceleration.

        if (data.count() > 0) {
            // transform into double array
            JavaRDD<double[]> doubles = DataManager.toDouble(data);
            // transform into vector without timestamp
            JavaRDD<Vector> vectors = doubles.map(Vectors::dense);
            // data with only timestamp and acc
            JavaRDD<long[]> timestamp = DataManager.withTimestamp(data);

            ////////////////////////////////////////
            // extract features from this windows //
            ////////////////////////////////////////
            ExtractFeature extractFeature = new ExtractFeature(vectors);

            // the average acceleration
            double[] mean = extractFeature.computeAvgAcc();

            // the variance
            double[] variance = extractFeature.computeVariance();

            // the average absolute difference
            double[] avgAbsDiff = computeAvgAbsDifference(doubles, mean);

            // the average resultant acceleration
            double resultant = computeResultantAcc(doubles);

            // the average time between peaks
            double avgTimePeak = extractFeature.computeAvgTimeBetweenPeak(timestamp);

            features = new double[]{
                    mean[0],
                    mean[1],
                    mean[2],
                    variance[0],
                    variance[1],
                    variance[2],
                    avgAbsDiff[0],
                    avgAbsDiff[1],
                    avgAbsDiff[2],
                    resultant,
                    avgTimePeak};
        }
        return Vectors.dense(features);
}