import org.apache.spark.mllib.linalg.Matrix; //导入依赖的package包/类
/**
 * This method performs PCA for a given dataset.
 *
 * @param workflowID The workflow ID associated with this dataset
 * @param dataSet DataSet on which PCA is performing (in JavaRDD<Vector> format)
 * @param noComponentsRetained Number of singular values retained after PCA operation
 * @throws DecompositionException
 */
@Override
public void fitPCA(String workflowID, JavaRDD<Vector> dataSet, int noComponentsRetained)
        throws DecompositionException {
    if (workflowID == null || workflowID.length() == 0) {
        throw new DecompositionException("Argument: workflowId is either null or empty");
    }

    if (dataSet == null) {
        throw new DecompositionException("Argument: dataSet is null for workflow Id: " + workflowID);
    }

    if (noComponentsRetained <= 0) {
        throw new DecompositionException(
                "Argument: noComponentsRetained is either zero or negative for workflow ID: " + workflowID);
    }

    RowMatrix dataMatrix = new RowMatrix(dataSet.rdd());
    Matrix pcaTransFormedMatrix = dataMatrix.computePrincipalComponents(noComponentsRetained);
    SparkDecompositionServiceUtil.saveMatrix(workflowID, pcaTransFormedMatrix);
}
 

import org.apache.spark.mllib.linalg.Matrix; //导入依赖的package包/类
/**
 * This method transforms a given dataset using pre-calculated PCA.
 *
 * @param workflowID The workflow ID associated with this dataset
 * @param dataSet DataSet on which PCA is performing (in JavaRDD<Vector> format
 * @return Transformed dataset in JavaRDD<Vector> format
 * @throws DecompositionException
 */
@Override
public JavaRDD<Vector> transformPCA(String workflowID, JavaRDD<Vector> dataSet) throws DecompositionException {
    if (workflowID == null || workflowID.length() == 0) {
        throw new DecompositionException("Argument: workflowId is either null or empty");
    }

    if (dataSet == null) {
        throw new DecompositionException("Argument: dataSet is null for workflow ID: " + workflowID);
    }

    RowMatrix dataMatrix = new RowMatrix(dataSet.rdd());
    Matrix principleComponents = SparkDecompositionServiceUtil.loadMatrix(workflowID);
    if (principleComponents == null) {
        throw new DecompositionException("PCA matrix is null for workflow ID: " + workflowID);
    }
    RowMatrix projectedMatrix = dataMatrix.multiply(principleComponents);
    return projectedMatrix.rows().toJavaRDD();
}
 

import org.apache.spark.mllib.linalg.Matrix; //导入依赖的package包/类
private boolean matrixEquals(Matrix mlMatrix, INDArray indMatrix, Double eps) {
    final int mlRows = mlMatrix.numRows();
    final int mlCols = mlMatrix.numCols();
    final int indRows = indMatrix.rows();
    final int indCols = indMatrix.columns();

    if (mlRows != indRows)
        return false;
    if (mlCols != indCols)
        return false;

    for (int i = 0; i < mlRows; i++) {
        for (int j = 0; j < mlCols; j++) {
            double delta = Math.abs(mlMatrix.apply(i, j) - indMatrix.getDouble(i, j));
            if (delta > eps)
                return false;
        }
    }
    return true;
}
 

import org.apache.spark.mllib.linalg.Matrix; //导入依赖的package包/类
/**
 * Convert a local (not distributed) Spark Matrix to an Apache Commons matrix.
 *
 * @param r Never {@code null}
 * @return Not {@code null}
 */
public static RealMatrix convertSparkMatrixToRealMatrix(final Matrix r){
    final RealMatrix result = new Array2DRowRealMatrix(r.numRows(), r.numCols());
    final double [] columnMajorMat = r.toArray();
    for (int i = 0; i < r.numRows(); i++) {
        result.setRow(i, Arrays.copyOfRange(columnMajorMat, i * r.numCols(), i * r.numCols() + r.numCols()) );
    }
    return result;
}
 

import org.apache.spark.mllib.linalg.Matrix; //导入依赖的package包/类
/**
 * Tangent normalize given the raw PoN data using Spark:  the code here is a little more complex for optimization purposes.
 *
 *  Please see notes in docs/PoN ...
 *
 *  Ahat^T = (C^T P^T) A^T
 *  Therefore, C^T is the RowMatrix
 *
 *  pinv: P
 *  panel: A
 *  projection: Ahat
 *  cases: C
 *  betahat: C^T P^T
 *  tangentNormalizedCounts: C - Ahat
 */
private static PCATangentNormalizationResult tangentNormalizeSpark(final ReadCountCollection targetFactorNormalizedCounts,
                                                                   final RealMatrix reducedPanelCounts,
                                                                   final RealMatrix reducedPanelPInvCounts,
                                                                   final CaseToPoNTargetMapper targetMapper,
                                                                   final RealMatrix tangentNormalizationInputCounts,
                                                                   final JavaSparkContext ctx) {
    // Make the C^T a distributed matrix (RowMatrix)
    final RowMatrix caseTDistMat = SparkConverter.convertRealMatrixToSparkRowMatrix(
            ctx, tangentNormalizationInputCounts.transpose(), TN_NUM_SLICES_SPARK);

    // Spark local matrices (transposed)
    final Matrix pinvTLocalMat = new DenseMatrix(
            reducedPanelPInvCounts.getRowDimension(), reducedPanelPInvCounts.getColumnDimension(),
            Doubles.concat(reducedPanelPInvCounts.getData()), true).transpose();
    final Matrix panelTLocalMat = new DenseMatrix(
            reducedPanelCounts.getRowDimension(), reducedPanelCounts.getColumnDimension(),
            Doubles.concat(reducedPanelCounts.getData()), true).transpose();

    // Calculate the projection transpose in a distributed matrix, then convert to Apache Commons matrix (not transposed)
    final RowMatrix betahatDistMat = caseTDistMat.multiply(pinvTLocalMat);
    final RowMatrix projectionTDistMat = betahatDistMat.multiply(panelTLocalMat);
    final RealMatrix projection = SparkConverter.convertSparkRowMatrixToRealMatrix(
            projectionTDistMat, tangentNormalizationInputCounts.transpose().getRowDimension()).transpose();

    // Subtract the projection from the cases
    final RealMatrix tangentNormalizedCounts = tangentNormalizationInputCounts.subtract(projection);

    // Construct the result object and return it with the correct targets.
    final ReadCountCollection tangentNormalized = targetMapper.fromPoNtoCaseCountCollection(
            tangentNormalizedCounts, targetFactorNormalizedCounts.columnNames());
    final ReadCountCollection preTangentNormalized = targetMapper.fromPoNtoCaseCountCollection(
            tangentNormalizationInputCounts, targetFactorNormalizedCounts.columnNames());
    final RealMatrix tangentBetaHats = SparkConverter.convertSparkRowMatrixToRealMatrix(
            betahatDistMat, tangentNormalizedCounts.getColumnDimension());

    return new PCATangentNormalizationResult(tangentNormalized, preTangentNormalized, tangentBetaHats.transpose(), targetFactorNormalizedCounts);
}
 

import org.apache.spark.mllib.linalg.Matrix; //导入依赖的package包/类
/**
 * This method is used to visualize a given dataset associated with a given workflowID.
 *
 * @param workflowID The workflow ID associated with this dataset
 * @response Name of the response variable
 * @return Fits two principle components of the transformed dataset
 * @throws DecompositionException
 */
@Override
public List<PCAResult> visualizePCA(String workflowID, String response) throws DecompositionException {
    if (workflowID == null || workflowID.length() == 0) {
        throw new DecompositionException("Argument: workflowId is either null or empty");
    }

    if (response == null || response.length() == 0) {
        throw new DecompositionException("Argument: response is either null or empty for workflow ID: "
                + workflowID);
    }

    JavaRDD<LabeledPoint> dataSet = SparkDecompositionServiceUtil.getSamplePoints(workflowID, response);
    JavaRDD<Double> labelsRDD = dataSet.map(new DataPointToResponseMapper());
    JavaRDD<Vector> features = dataSet.map(new DataPointToFeatureMapper());
    RowMatrix dataMatrix = new RowMatrix(features.rdd());

    // extracting first two principle components for visualization
    Matrix pcaTransFormedMatrix = dataMatrix.computePrincipalComponents(2);
    RowMatrix projectedMatrix = dataMatrix.multiply(pcaTransFormedMatrix);

    List<Vector> pcaDataPoints = projectedMatrix.rows().toJavaRDD().toArray();
    List<Double> labels = labelsRDD.toArray();

    List<PCAResult> pcaResults = new ArrayList<PCAResult>();
    for (int i = 0; i < pcaDataPoints.size(); i++) {
        double[] pcaData = pcaDataPoints.get(i).toArray();
        double label = labels.get(i);
        pcaResults.add(new PCAResult(pcaData[0], pcaData[1], label));
    }
    return pcaResults;
}
 

import org.apache.spark.mllib.linalg.Matrix; //导入依赖的package包/类
/**
 * Save serialized matrix in hard disk.
 *
 * @param workflowID The workflow ID associated with this dataset
 * @param matrix The matrix which is going to save in the disk
 * @throws DecompositionException
 */
public static void saveMatrix(String workflowID, Matrix matrix) throws DecompositionException {
    FileOutputStream fileOutStream = null;
    ObjectOutputStream matrixOutStream = null;
    try {
        String fullPath = buildPCAMatrixPath(workflowID);

        // if file is already exists, delete it
        File currentMatrix = new File(fullPath);
        if (currentMatrix.exists()) {
            boolean isSuccess = currentMatrix.delete();
            if (!isSuccess) {
                throw new DecompositionException("An error occurred while deleting matrix, in workflow : "
                        + workflowID);
            }
        }

        // serialize and save matrix object into disk
        fileOutStream = new FileOutputStream(fullPath);
        matrixOutStream = new ObjectOutputStream(fileOutStream);
        matrixOutStream.writeObject(matrix);

    } catch (IOException ex) {
        throw new DecompositionException("An error occurred while saving a matrix: " + ex.getMessage(), ex);
    } finally {
        closeResource(fileOutStream);
        closeResource(matrixOutStream);
    }
}
 

import org.apache.spark.mllib.linalg.Matrix; //导入依赖的package包/类
/**
 * Convert an ndarray to a matrix.
 * Note that the matrix will be con
 * @param arr the array
 * @return an mllib vector
 */
public static Matrix toMatrix(INDArray arr) {
    if (!arr.isMatrix()) {
        throw new IllegalArgumentException("passed in array must be a matrix");
    }

    // if arr is a view - we have to dup anyway
    if (arr.isView()) {
        return Matrices.dense(arr.rows(), arr.columns(), arr.dup('f').data().asDouble());
    } else // if not a view - we must ensure data is F ordered
        return Matrices.dense(arr.rows(), arr.columns(),
                        arr.ordering() == 'f' ? arr.data().asDouble() : arr.dup('f').data().asDouble());
}
 

import org.apache.spark.mllib.linalg.Matrix; //导入依赖的package包/类
@Test
public void testINDtoMLMatrix() {
    INDArray matIND = Nd4j.rand(23, 100);

    Matrix matMl = MLLibUtil.toMatrix(matIND);

    assertTrue(matrixEquals(matMl, matIND, 0.01));
}
 

import org.apache.spark.mllib.linalg.Matrix; //导入依赖的package包/类
@Test
public void testMltoINDMatrix() {
    Matrix matMl = Matrices.randn(23, 100, new Random(3949955));

    INDArray matIND = MLLibUtil.toMatrix(matMl);
    log.info("matrix shape: {}", Arrays.toString(matIND.shapeInfoDataBuffer().asInt()));

    assertTrue(matrixEquals(matMl, matIND, 0.01));
}
 

import org.apache.spark.mllib.linalg.Matrix; //导入依赖的package包/类
/**
 * Predict the given feature matrix
 *
 * @param features the given feature matrix
 * @return the predictions
 */
public Matrix predict(Matrix features) {
    return MLLibUtil.toMatrix(network.output(MLLibUtil.toMatrix(features)));
}
 

import org.apache.spark.mllib.linalg.Matrix; //导入依赖的package包/类
/**
 * Predict the given feature matrix
 * @param features the given feature matrix
 * @return the predictions
 */
public Matrix predict(Matrix features) {
    return MLLibUtil.toMatrix(layer.activate(MLLibUtil.toMatrix(features)));
}