实战案例 | 使用机器学习和大数据预测心脏病

SparkConfAndCtxBuilder ctxBuilder = new SparkConfAndCtxBuilder()；
JavaSparkContext jctx = ctxBuilder.loadSimpleSparkContext("Heart Disease Detection App"， "local")；

//读取数据到RDD，数据是逐行分割的字符串格式
JavaRDD<String> dsLines = jctx.textFile(trainDataLoc)；
        // 使用适配器类解析每个文本行
        // 现在数据已经被转换成模型需要的格式了
JavaRDD<LabeledPoint> _modelTrainData = dsLines.map(new DataToModelAdapterMapper())；

    //我们需要的模型在被数据训练
        //你可以替代下面的代码，来尝试使用决策树模型，并比较返回数据的精度

NaiveBayesModel _model = NaiveBayes.train(_modelTrainData.rdd())；
_model.save(jctx.sc()， modelStorageLoc)；

ctxBuilder.closeCtx()；

public LabeledPoint call(String dataRow) throws Exception {
    //用一个很大的值替代空的数据点，来避免不必要的空数据点
    String newLine = dataRow.replaceAll("\？"， "99.0")；
    String[] tokens = newLine.split("，")；

    System.out.println("tokens count ： " + tokens.length)；
    // 上一个token有被训练模型使用的实际预测值
    Integer lastToken = Integer.parseInt(tokens[13])；

    double[] featuresDblArr = new double[13]；
    for(int i = 0； i < 13； i++) {
        featuresDblArr[i] = Double.parseDouble(tokens[i])；
    }
    // 构建特征向量
    Vector featuresVector = new DenseVector(featuresDblArr)；

    Double classValue = 0.0；
    if(lastToken.intValue() > 0 ) classValue = 1.0；

    LabeledPoint _lp = new LabeledPoint(classValue， featuresVector)；

    return _lp；
}

String schemaString = "age sex cp trestbps chol fbs restecg thalach exang oldpeak slope ca thal num"；
List<StructField> fields = new ArrayList<>()；
for (String fieldName ： schemaString.split(" ")) {
    fields.add(DataTypes.createStructField(fieldName， DataTypes.StringType， true))；
}

StructType schema = DataTypes.createStructType(fields)；
JavaRDD<Row> rowRdd = rows.map(new Function<String， Row>() {

    @Override
    public Row call(String record) throws Exception {
        String[] fields = record.split("，")；
        return RowFactory.create(fields[0]，fields[1]，fields[2]，fields[3]，fields[4]，fields[5]，fields[6]，fields[7]，fields[8]，fields[9]，fields[10]，fields[11]，fields[12]，fields[13])；
    }
})；

DataFrame df = sqlCtx.createDataFrame(rowRdd， schema)；
df.registerTempTable("heartDisData")；

DataFrame results = sqlCtx.sql("select min(age) from heartDisData")；

JavaRDD<String> jrdd = results.javaRDD().map(new Function<Row， String>() {

    @Override
    public String call(Row arg0) throws Exception {
        return arg0.toString()；
    }

})；

List<String> rstList = jrdd.collect()；

for (String rStr ： rstList) {
    System.out.println(" Minimum Age ： " + rStr)；
}

NaiveBayesModel _model = NaiveBayesModel.load(<Spark Context>， <Model Storage Location>)；

SparkConfAndCtxBuilder ctxBuilder = new SparkConfAndCtxBuilder()；
JavaSparkContext jctx = ctxBuilder.loadSimpleSparkContext("Heart Disease Detection App"， "local")；

JavaRDD<String> dsLines = jctx.textFile(testDataLoc)；
JavaRDD<Vector> fRdd = dsLines.map(new TestDataToFeatureVectorMapper())；

NaiveBayesModel _model = NaiveBayesModel.load(jctx.sc()， modelStorageLoc)；

JavaRDD<Double> predictedResults = _model.predict(fRdd)；
List<Double> prl = predictedResults.collect()；
for (Double pr ： prl) {
    System.out.println("Predicted Value ： " + pr)；
}