Spark大数据技术与应用（第3版）（微课版）课件 项目8 饮用水源合格性预测-Spark MLlib机器学习算法库V1.0

饮用水源合格性预测——

SparkMLlib机器学习算法库项目背景水是生命之源，是人类生存和发展的基础。它是生态系统中不可或缺的组成部分，对于维系地球上的生命以及农业、工业、城市发展等方面都具有至关重要的作用。饮用水安全直接关系到人类的健康和生存。水中的污染物和有害物质可能对人体造成严重的危害，如引发肠道疾病、皮肤疾病等。因此，确保饮用水安全是保障人类健康的重要前提。随着工业化、城市化进程的加快，农业面源污染、生活污水排放等问题也日益严重，进一步加剧了水源地的富营养化问题。这些因素共同作用，使得饮用水源合格性预测成为当前亟待实现的重要任务。项目背景现有一份饮用水源的指标检测数据，数据由约8000条模拟数据记录组成，具体字段如下。字段名称说明字段名称说明字段名称说明aluminium铝copper铜mercury汞ammonia氨fluoride氟化物perchlorate高氯酸盐arsenic砷bacteria细菌radium镭barium钡viruses病毒selenium硒cadmium镉lead铅silver银chloramine氯胺nitrates硝酸盐uranium铀chromium铬nitrites亚硝酸盐is_safe是否合格机器学习算法简介使用MLlib机器学习算法库项目实施机器学习概念机器学习指的是让机器能像人一样具备学习、理解、认识的能力。机器学习主要研究如何在经验学习中改善具体算法的性能。机器学习的过程步骤：首先，通过计算机对历史数据的规律或以往经验进行学习，并构建算法模型。然后，对模型进行评估：如果评估的性能达到要求，该模型即可用于测试其他数据；如果达不到要求，则需要调整算法，重新建立模型，并再次进行评估。最后，直到获得达到要求的算法模型才停止循环。机器学习概念机器学习可以分为监督学习、无监督学习、半监督学习3种。监督学习的训练数据是有标签的，即已经能够确定所给数据集的类别。半监督学习针对的是数据量非常大但是标签数据很少或者标签数据不易获取的情况。无监督学习与监督学习相反，训练数据完全没有标签，只能依靠数据间的相似性进行分类，如广泛使用的K-Means算法就属于无监督学习算法。机器学习算法机器学习是很多重要学科或领域的支撑，其算法众多，主要有以下几种。回归算法分类算法聚类算法推荐算法降维算法机器学习算法简介使用MLlib机器学习算法库项目实施MLlib简介SparkMLlib旨在简化机器学习的工程实践工作，通过分布式并行计算进行海量数据的迭代计算，从而实现模型。MLlib对数据进行处理的速度远超普通数据处理引擎，显著提升运行性能。MLlib由多种通用的机器学习算法和工具组成，主要包括以下内容。算法：常用的机器学习算法，如分类、回归、聚类和协同过滤等算法。特征化：对特征进行提取、转化、降维和选择。管道（Pipeline）：用于构建、评估和调整机器学习工作流的工具。持久化：保存和加载算法、模型及管道的状态。实用工具：线性代数、统计、数据处理等工具。MLlib发展历史SparkMLlib历史悠久，1.0以前版本提供的算法均基于RDD实现。SparkMLlib发展历史主要如下。0.8版本时，MLlib算法库被加入Spark，但只支持Java和Scala两种语言。1.0版本时，SparkMLlib支持使用Python语言。自1.2版本起，SparkMLlib分为以下两个包。spark.mllib包提供基于RDD的算法API。spark.ml包提供基于DataFrame的算法API，用于构建机器学习工作流。该包弥补了原始MLlib库的不足，提供基于DataFrame的机器学习工作流式API套件。1.2以后版本，持续增加和改进MLlib中的算法。从Spark2.0起，基于RDD的API进入维护模式，不再新增特性。Spark官方更推荐使用spark.ml包。若新算法符合机器学习管道概念，可纳入spark.ml包。数据类型MLlib中常见的基本数据类型：数据类型描述LabeledPoint监督学习算法的数据对象，用于表示带标签的数据，包含两个部分。第一部分：数据的类别标签label，由一个浮点数表示；第二部分：特征向量feature，向量通常是Vector向量数据类型，向量中的值是Double类型。数据类型定义在mllib.regression中。对于二类分类，标签应该是0.0（负）或1.0（正）。对于多分类，标签是从0.0开始的索引，即0.0、1.0、2.0等。Rating对产品的评分，用于ALS推荐算法。ALS协同过滤推荐算法所使用的类型，存放在mllib.recommendation中，包括用户ID、商品ID和rating（评分）3个部分，用户ID和商品ID要求为Int类型数据。数据类型数据类型描述Vector数据向量，包括稀疏向量和稠密向量。稠密向量存储向量的每一个值。稀疏向量存储非零向量，由索引数组和值数组这两个并行数组支持。例如，对于向量(1,0,3)，稠密向量格式表示为[1.0,0.0,3.0]；稀疏向量则表示为(3,[0,2],[1,3])，其中3是向量的大小，[0,2]为索引数组，[1,3]为值数组。在使用算法包时，很多算法都要求将数据转化为向量，Vector向量数据类型可以由mllib.linalg.Vectors创建。特征提取TF-IDF算法计算TF-IDF值的具体步骤如下：对于给定的语料库，首先计算该文档中每个词语的词频TF：计算每个词语在整个语料库中的逆文档频率IDF：将TF和IDF相乘，得到该文档中每个词语的TF-IDF值：根据TF-IDF值的大小进行排序，从而确定关键词。TF-IDF算法在SparkMLlib中，TF-IDF算法的实现通过HashingTF和IDF。HashingTF从一个文档中计算出给定大小的词频向量，并且通过哈希法排列词向量的顺序，使词与向量能一一对应。IDF则可以计算逆文档频率，需要调用fit()方法获取一个IDFModel，表示语料库的逆文档频率，再通过IDFModel的transform()方法将TF向量转换为IDF向量。TF-IDF算法以tf-idf.txt数据为例，spark.mllib包中的TF-IDF算法实现。首先需要读取文本数据，对每一个文档，将其分割成单词序列。之后使用HashingTF()方法将每一个单词序列转换为特征向量。最后再调用IDF()方法重新调整特征向量，得到文档转化后的特征向量。HiIheardaboutSparkIwishJavacouldusecaseclassesLogisticregressionmodelsareneatimportorg.apache.spark.mllib.feature.{HashingTF,IDF}valdocuments=sc.textFile("/tipdm/data/tf-idf.txt").map(_.split("").toSeq)valhashingTF=newHashingTF()valtf=hashingTF.transform(documents)tf.cache()validf=newIDF().fit(tf)valtfidf=idf.transform(tf)tfidf.collect().foreach(println)TF-IDF算法结果如图所示，其中每一条输出的第一个值为默认的哈希表的分桶个数（220）。前一个列表是各个单词在特征向量中的索引列表。后一个列表则记录了前一个列表中每个单词在该单词序列（文档）中的TF-IDF值。TF-IDF算法spark.ml包中同样提供了TF-IDF算法，与spark.mllib包中提供的算法不同的是，spark.mllib包中是通过数据创建RDD。spark.ml包中则是通过数据创建DataFrame，之后再针对DataFrame去实现TF-IDF算法，spark.ml包中的TF-IDF算法。TF-IDF算法结果如图所示，features列便是经过TF-IDF算法转换后的特征向量。特征向量中第一个值表示特征维度，后续列表中的两部分分别对应着单词的索引与TF-IDF值。在spark.ml包中TF-IDF算法实现的代码中，前两行数据分别有5个和7个单词;而在运行结果图中，前两行的列表中分别只有4个和6个值。这可能是因为哈希函数将不同的单词映射到了同一位置（索引），或者是将单词的索引映射到了超出特征向量维度的位置。该问题可以通过调整特征维度来解决。特征提取Word2Vec算法Word2Vec是NLP（自然语言处理）领域的重要算法，它能将词语转换为一组表示语义信息的向量。Word2Vec能够捕捉到词语之间的复杂关系，如语义相似性、词的上下文关联等，在执行NLP任务时能更好地理解文本内容。Word2Vec算法使用的训练集是语料库，其中每个文档不含标点，以空格断句，表示为一个单词序列。importorg.apache.spark.mllib.feature.{Word2Vec,Word2VecModel}valinput=sc.textFile("/tipdm/data/w2v.txt").map(line=>line.split("").toSeq)valword2vec=newWord2Vec().setMinCount(0)valmodel=word2vec.fit(input)valsynonyms=model.findSynonyms("spark",5)for((synonym,cosineSimilarity)<-synonyms){println(s"$synonym$cosineSimilarity")}Word2Vec算法以w2v.txt数据为例，一行表示一个文档，整份数据表示一个语料库。通过spark.mllib包中的Word2Vec算法寻找与“spark”语义相同的5个词，并输出其相似度。sparkisafastandgeneralengineforlargescaledataprocessingword2vecisapopularalgorithmforlearningwordembeddingsmachinelearningalgorithmsrequirelargeamountsoftrainingdatanaturallanguageprocessinginvolvesunderstandinghumanlanguageWord2Vec算法spark.ml包中的Word2Vec算法实现与spark.mllib包中的Word2Vec算法实现原理相同。importorg.apache.spark.ml.feature.Word2VecvaldocumentDF=spark.createDataFrame(Seq("sparkisafastandgeneralengineforlargescaledataprocessing".split(""),"word2vecisapopularalgorithmforlearningwordembeddings".split(""),"machinelearningalgorithmsrequirelargeamountsoftrainingdata".split(""),"naturallanguageprocessinginvolvesunderstandinghumanlanguage".split("")valword2Vec=newWord2Vec().setInputCol("text").setOutputCol("result").setMinCount(0)valmodel=word2Vec.fit(documentDF)model.findSynonyms("spark",5).show()特征提取统计最大值、最小值、均值、方差和相关系数spark.mllib.stat包中提供的统计方法：方法描述count()计数max()/min()最大值/最小值mean()均值normL1()/normL2()L1范数/L2范数variance()方差统计最大值、最小值、均值、方差和相关系数方法描述Statistics.corr(rdd,method)计算单个RDD[Vector]中元素之间的相关性，返回一个相关系数矩阵，method默认为pearsonStatistics.corr(rdd1,rdd2,method)计算两个RDD[Double]之间的相关性，返回一个相关系数，method默认为pearsonStatistics.chiSqTest（rdd）计算输入数据RDD[LabeledPoint]中每个特征与标签之间的卡方统计量，检验特征与标签是否独立（独立性检验）importorg.apache.spark.mllib.linalg.Vectorsimportorg.apache.spark.mllib.stat.Statisticsvaldata=sc.textFile("/tipdm/data/stat.txt").map(_.split("")).map(f=>f.map(f=>f.toDouble))valdata1=data.map(f=>Vectors.dense(f))valstat1=Statistics.colStats(data1)stat1.meanstat1.variancevalcorr1=Statistics.corr(data1,"pearson")统计最大值、最小值、均值、方差和相关系数以数据文件stat.txt为例，读取stat.txt文件的数据并创建RDD，然后统计RDD数据的均值、方差和相关系数矩阵。1.02.03.04.05.06.07.01.05.09.03.05.06.03.01.03.01.01.05.06.0统计最大值、最小值、均值、方差和相关系数同样地，SparkMLlib的spark.ml.stat包中也提供了一些常见的统计方法，可以对DataFrame对象进行处理。以统计DataFrame中的数据的相关系数矩阵和独立性检验的方法为例。特征提取数据特征处理方法为避免数据字段的量纲和量级的不同对模型效果造成不良影响，经常需要对数据进行标准化或归一化。经过数据标准化或归一化后，算法效果在一定程度上会提升。SparkMLlib中提供的常见的数据处理的方法包括Normalizer()、StandardScaler()和MinMaxScaler()方法，其中spark.mllib包中只有前两种数据处理方法，spark.ml包中则含有3种。接下来将使用spark.ml包中的方法进行数据特征处理。数据特征处理方法Normalizer()、StandardScaler()和MinMaxScaler()方法处理的均为Vector类型数据，因此需要先将数据转换为Vector类型。将一个集合序列转化为RDD，再由RDD数据创建DataFrame，最后将DataFrame中的数据类型转换成Vector类型。importorg.apache.spark.ml.linalg.VectorsvaldataFrame=spark.createDataFrame(Seq((0,Vectors.dense(1.0,0.5,-1.0)),(1,Vectors.dense(2.0,1.0,1.0)),(2,Vectors.dense(4.0,10.0,2.0)))).toDF("id","features")Normalizer()

方法将数据的类型转换成Vector类型后，使用Normalizer()进行数据归一化的介绍及使用方法如下。Normalizer()方法本质上是一个转换器，它可以将多行向量输入转化为统一的形式。Normalizer()方法的作用范围是每一行，使每一个行向量的范数变换为一个单位范数。参数setP用于指定正则化中使用的p-norm，默认值为2。对Vector类型的DataFrame数据进行Normalizer归一化操作。importorg.apache.spark.ml.feature.Normalizervalnormalizer=newNormalizer().setInputCol("features").setOutputCol("normFeatures").setP(1.0)vall1NormData=normalizer.transform(dataFrame)l1NormData.show()setInputCol("features")设置了Normalizer归一化的输入数据，setOutputCol("normFeatures")设置了Normalizer归一化后输出的数据作为DataFrame中的normFeatures列。StandardScaler()方法StandardScaler()方法处理的对象是列，即每一维特征，将其标准化为单位标准差、0均值或0均值单位标准差。StandardScaler()方法有两个参数可以设置，说明如下。withStd：true或false，默认为true，该参数表示是否将数据标准化到单位标准差。withMean：true或false，默认为false，该参数表示是否变换为0均值，将返回一个稠密输出，因此不适用于稀疏输入。StandardScaler()方法进行StandardScaler标准化需要获取数据每一维的均值和标准差，并以此缩放每一维特征。对Vector类型的DataFrame数据进行StandardScaler标准化。结果如图所示，每列数据均按比例进行缩放，features列为未进行StandardScaler标准化的原数据列，scaledFeatures列为进行StandardScaler标准化后的数据列。importorg.apache.spark.ml.feature.StandardScalervalscaler=newStandardScaler().setInputCol("features").setOutputCol("scaledFeatures").setWithStd(true).setWithMean(false)valscalerModel=scaler.fit(dataFrame)valscaledData=scalerModel.transform(dataFrame)scaledData.show(false)MinMaxScaler()方法MinMaxScaler()为常用的最小值——最大值归一化方法，这个方法也针对每一维特征进行处理，将每一维特征线性地映射到指定的区间中，通常是[0,1]。MinMaxScaler()方法有两个参数可以设置，说明如下：min：默认为0，指定区间的下限。max：默认为1，指定区间的上限。对Vector类型的DataFrame数据进行MinMaxScaler归一化，将每一列数据映射在区间[0,1]中。importorg.apache.spark.ml.feature.MinMaxScalervalscaler=newMinMaxScaler().setInputCol("features").setOutputCol("scaledFeatures")valscalerModel=scaler.fit(dataFrame)valscaledData=scalerModel.transform(dataFrame)scaledData.show(false)MinMaxScaler()方法结果如图所示，features列为未进行MinMaxScaler归一化的原数据列，scaledFeatures列为进行MinMaxScaler归一化后的数据列，每一列的数据均被映射到区间[0,1]。回归线性回归线性回归通过一组线性组合预测输出值。在SparkMLlib中可以通过spark.ml.regression包中的LinearRegression类实现线性回归算法，该算法主要的参数包括最大迭代次数setMaxIter()、正则化参数setRegParam()和弹性网络混合参数setElasticNetParam()。以Spark安装目录下的data/mllib/sample_linear_regression_data.txt文件作为输入数据。数据为LIBSVM格式，即第一列为预测值或标签，其他列为特征列，特征列格式为“索引:值”，其中“索引”是表示特征的列号或维度的从1开始的整数，“值”则是表示该特征的具体数值。-9.4900098788245481:0.45512736006573622:0.366446943519690873:-0.382561089334680474:-0.44584301985172675:0.331097903589147266:0.80674452934435657:-0.26243417317738878:-0.448503861116595249:-0.0726928483816933210:0.5658035575800715线性回归将数据上传至HDFS的/tipdm/data目录下，并调用spark.ml.regression包中的LinearRegression类实现线性回归算法。importorg.apache.spark.ml.regression.{LinearRegression,LinearRegressionModel}valdata=spark.read.format("libsvm").load("/tipdm/data/sample_linear_regression_data.txt")valArray(train,test)=ModelData.randomSplit(Array(0.7,0.3))vallr=newLinearRegression().setMaxIter(10).setRegParam(0.3).setElasticNetParam(0.8)vallrModel=lr.fit(train)valpre=lrModel.transform(test)lrModel.save("myModelPath")valmodel=LinearRegressionModel.load("myModelPath")回归逻辑回归逻辑回归是一种二分类的回归算法，预测的值为新点属于每个类的概率，将概率大于等于阈值的分到一个类，小于阈值的分到另一个类。在SparkMLlib中可以通过spark.ml.classification包中的LogisticRegression类实现逻辑回归算法，逻辑回归算法的主要参数与使用方法与线性回归类似。分类朴素贝叶斯朴素贝叶斯是一种十分简单的分类算法。朴素贝叶斯对于给出的特定features，求解在此项出现条件下各个类别出现的概率，并将其归类为概率最大的类别。在SparkMLlib中，可以通过调用spark.ml.classification包下的NaiveBayes类实现朴素贝叶斯算法，有多分类和二分类两种方式。朴素贝叶斯以根据天气的情况判断是否出去打球为例，因为朴素贝叶斯算法要求输入为数值类型的数据，所以将字符类型数据转化为数值类型数据，并保存为weather_data.txt文件。朴素贝叶斯算法要求输入为数值类型的数据，因此将字符类型数据转化为数值类型数据，并保存为weather_data.txt文件。第1列为是否打球的标签，后面依次为天气、温度、湿度、是否刮风的特征列。是否打球天气温度湿度是否刮风否晴天较高湿是否晴天较高湿否01:02:13:14:101:02:13:14:0朴素贝叶斯将数据上传至HDFS的/tipdm/data目录下，并调用朴素贝叶斯算法构建分类模型。importorg.apache.spark.ml.classification.NaiveBayesimportorg.apache.spark.ml.evaluation.MulticlassClassificationEvaluatorvaldata=spark.read.format("libsvm").load("/tipdm/data/weather_data.txt")valArray(trainingData,testData)=data.randomSplit(Array(0.6,0.4))valmodel=newNaiveBayes().fit(trainingData)valpredictions=model.transform(testData)predictions.show()valevaluator=newMulticlassClassificationEvaluator().setLabelCol("label").setPredictionCol("prediction").setMetricName("accuracy")valaccuracy=evaluator.evaluate(predictions)println(s"Testsetaccuracy=$accuracy")分类决策树决策树是分类和回归的常用算法，因为决策树容易处理类别特征，所以比较适合处理多分类的问题。SparkMLlib支持二分类和多分类的决策树。决策树以节点树的形式表示，每个节点代表一个向量，向量的不同特征值会使节点有多条指向下个节点的边，最底层的叶子节点为预测的结果，可以是分类的特征，也可以是连续的特征。每个节点的选择都遵循某一种使模型更加优化的算法，如基于信息增益最大的方法。在SparkMLlib中，可以调用spark.ml.classification包下的DecisionTreeClassifier类实现决策树分类算法。聚类聚类是一种无监督学习的方法，用于将高度相似的数据分到一类中。聚类没有类别标签，仅根据数据相似度进行分类，因此聚类通常用于数据探索、异常检测，也用于一般数据的分群。聚类的方法有很多，计算相似度的方法也有很多，K-Means算法是较常使用的一种算法。在SparkMLlib中K-Means算法可以通过调用spark.ml.clustering算法包下的KMeans类实现。以Spark安装目录下的data/mllib/sample_kmeans_data.txt数据为例，将数据上传至HDFS的/tipdm/data目录下，并通过K-Means算法实现聚类。01:0.02:0.03:0.011:0.12:0.13:0.121:0.22:0.23:0.2模型将前三条数据划分为类别“1”，将后三条数据划分为类别“0”。“silhouette”表示轮廓系数，取值范围为[-1,1]，值越大表示聚类效果越好，此处轮廓系数为0.999，说明模型效果非常好。聚类关联规则FP算法主要通过大量的客户购买历史数据生成频繁项集，设置支持度，筛选出符合支持度的频繁项集，根据频繁项集生成一些规则后，再通过置信度过滤出较有说服力的强关联规则。通过强关联规则即可完成推荐、分类等工作。在SparkMLlib中实现FP算法可以通过spark.ml.fpm包中的FPGrowth类。以餐饮企业的点餐数据为例，调用FP算法计算菜品之间的关联性。某餐饮企业的点餐数据保存在MySQL数据库中。序列时间订单号菜品ID菜品名称12014/8/2110118491健康麦香包关联规则将点餐数据中的特殊记录数据整理成关联规则模型所需的数据结构，从中抽取10个点餐订单作为事务数据集。为方便模型构建，将菜品ID简记为字母，如菜品ID为18491简记为a，菜品ID为8842简记为b等。提取简记后的菜品ID数据存为menu_orders.txt文件，将该文件数据作为关联规则的模型数据。订单号菜品ID简记后的菜品ID11849186938705ace288427794bd388428693bc418491884286937794abcdacebdbcabcd关联规则importorg.apache.spark.ml.fpm.FPGrowthvalrawDF=spark.read.text("/tipdm/data/menu_orders.txt")valtransactionsDF=rawDF.withColumn("items",split(trim(col("value")),""))valfpgrowth=newFPGrowth().setItemsCol("items").setMinSupport(0.5).setMinConfidence(0.6)valmodel=fpgrowth.fit(transactionsDF)model.freqItemsets.show()model.associationRules.show()model.transform(transactionsDF).show()推荐

在SparkMLlib中实现推荐算法可以通过spark.ml.recommendation包下的ALS类，该类可根据用户对各种产品的交互和评分推荐新产品，通过最小二乘法来求解模型。ALS有显式评分（默认）和隐式反馈两种方法，显式评分是指用户对商品有明确评分，预测结果也是评分，隐式反馈是指用户和产品的交互置信度，预测结果也是置信度。ALS模型的优化参数主要有4个，说明如下。rank：使用的特征向量的大小，更大的特征向量会产生更好的模型，但同时也需要花费更大的计算代价，默认为10。maxIter：算法最大迭代的次数，默认为10。regParam：正则化参数，默认为1.0。alpha：用于在隐式反馈ALS中计算置信度，默认为1.0。importorg.apache.spark.ml.recommendation.ALScaseclassRating(userId:Int,movieId:Int,rating:Float,timestamp:Long)defparseRating(str:String):Rating={valfields=str.split("::")assert(fields.size==4)Rating(fields(0).toInt,fields(1).toInt,fields(2).toFloat,fields(3).toLong)}

valratings=spark.read.textFile("/tipdm/data/sample_movielens_ratings.txt").map(parseRating).toDF()valArray(training,test)=ratings.randomSplit(Array(0.8,0.2))推荐

以Spark安装目录下的示例数据文件data\mllib\als\sample_movielens_ratings.txt为例。第1列为用户ID，第2列为商品ID，第3列为用户对商品的评分，第四列为用户评分的时间戳。将数据上传至HDFS的/tipdm/data目录下，调用ALS算法实现商品推荐。0::2::3::3::1::5::2::9::4::1424380312推荐

valals=newALS().setMaxIter(5).setRegParam(0.01).setUserCol("userId").setItemCol("movieId").setRatingCol("rating")valmodel=als.fit(training)valpredictions=model.transform(test)valuserRecs=model.recommendForAllUsers(10)valmovieRecs=model.recommendForAllItems(10)valusers=ratings.select(als.getUserCol).distinct().limit(3)valuserSubsetRecs=model.recommendForUserSubset(users,10)valmovies=ratings.select(als.getItemCol).distinct().limit(3)valmovieSubSetRecs=model.recommendForItemSubset(movies,10)MLlib中的模型评估

对于机器学习而言，无论使用哪种算法，模型评估都是非常重要的。通过模型评估可以知道模型的好坏，并预测分类结果的准确性，这有利于对模型进行修正。在SparkMLlib中可以通过spark.ml.evaluation包中的MulticlassClassificationEvaluator和RegressionEvaluator类实现模型评估。其中，MulticlassClassificationEvaluator类主要用于分类问题的模型评估，RegressionEvaluator主要用于回归问题的模型评估。对于分类问题，主要的评估指标有准确率、召回率、F值、ROC曲线等；对于回归问题，主要的评估指标有均方误差MSE、均方根误差RMSE、平均绝对误差MAE、决定系数R2等。在SparkMLlib中还有很多算法和数据处理方法，可以通过Spark官网进行进一步的学习。机器学习算法简介使用MLlib机器学习算法库项目实施读取数据本任务将读取饮用水源的指标检测数据waterQuality.csv，并对数据进行初步的数据清洗，包括缺失值处理和异常值处理等。首先需要将数据上传至HDFS的/tipdm/data目录下。接着，在IntelliJIDEA中创建一个Spark工程，导入Spark相关的开发依赖包，并创建一个water.scala类。在water.scala类的主函数中创建一个命名为Spark的SparkSession对象，最后读取HDFS数据创建DataFrame。探索性数据分析首先查看数据是否存在重复记录和空值，结果如图所示。探索水源合格与不合格的记录数量，结果如图所示。可以看到，在全部7996条数据中，“is_safe”字段标签为1（合格）的有912条，标签为0（不合格）的有7084条，水源不合格率高达88.6%，反映了对饮用水源合格性预测的重要性。探索性数据分析探索标签列（is_safe）与其他列的相关系数，需要先将所有字段转换为一个特征向量。然后计算该向量中的最后一个元素（is_safe）与其他元素的相关系数。为了更加直观地查看结果，可以逐行输出打印标签列与其他列的相关系数。valassembler=newVectorAssembler().setInputCols(waterData.columns).setOutputCol("features")valwaterDataWithFeatures=assembler.transform(waterData)valRow(corrMatrix:Matrix)=Correlation.corr(waterDataWithFeatures,"features").head()valnumRows=waterData.columns.lengthvalcorrArray=corrMatrix.toArrayfor(i<-0untilnumRows-1){valcorrelationValue=corrArray(i*numRows+(numRows-1))println(f"${waterData.columns.last}列与${waterData.columns(i)}列的相关系数：$correlationValue")}探索性数据分析除了标签列外的其他20个特征中，与is_safe列相关性较强的有aluminium、cadmium、chloramine、chromium、arsenic、silver等特征，相关系数均超过0.1。数据标准化在计算标签列与特征列的相关系数时，将所有字段合并为一个特征向量，但在进行数据标准化和模型训练时，标签列不应作为特征参与计算，因此在数据标准化前需重新构建特征列。valnewAssembler=newVectorAssembler().setInputCols(waterData.columns.init).setOutputCol("newFeatures")valscaler=newStandardScaler().setInputCol("newFeatures").setOutputCol("scaledFeatures").