大数据 | 实验四：并行化数据挖掘算法设计

文章目录

欢迎指正~

📚实验目的

机器学习和数据挖掘算法是大数据分析处理领域的重要内容，随着数据规模的不断扩大，设计面向大数据处理的并行化机器学习和数据挖掘算法越来越有必要。通过对并行化数据挖掘算法的实现，掌握并行化处理问题的分析方法和编程思想方法，能够根据实际情况定制并行化的算法解决问题。

📚实验平台

1）操作系统：Linux；
2）Hadoop 版本：3.2.2；
3）JDK 版本：1.8；
4）Java IDE：Eclipse 3.8；
5 ) Spark 版本：3.2.1。

📚实验内容

• 准备数据集（戳此下载），设计一种数据挖掘算法（聚类、分类、频繁项集挖掘或其他主题）对数据集进行信息提取，要求分别使用并行化和非并行化的方式实现该算法。实验环境可选择 Hadoop 或者 Spark，程序语言可选用 Java、Python、Scala 等。
• 在伪分布式环境下完成并行化算法的编写和测试，并在服务器集群进行提交运行。在单机环境下完成非并行化算法的编写和测试。
• 自行对比并行化和非并行化实现方法的数据挖掘结果，两种结果需完全一致。

📚实验步骤

🐇KNN介绍

• k近邻法（k-nearest neighbor，k-NN）是一种基本的分类和回归方法，是监督学习方法里的一种常用方法。k近邻算法假设给定一个训练数据集，其中的实例类别已定。分类时，对新的实例，根据其k个最近邻的训练实例类别，通过多数表决等方式进行预测。
• k近邻算法用一句通俗的古语来说就是：“物以类聚，人以群分”。如果要看一个实例的类别，那么可以看它附近是什么类别。
• k近邻法三要素：距离度量、 k值的选择和分类决策规则。常用的距离度量是欧氏距离。 k值小时，k近邻模型更复杂，容易发生过拟合；k值大时，k近邻模型更简单，又容易欠拟合。

🐇并行化

🥕在本地编写程序和调试

• Mapper：计算训练集中每个点与测试集中所有点的欧式距离，并将距离和对应的类标签输出。- 输入：<行偏移量,文件内容>,缓存的是test测试集，输入的是train训练集- 输出：<第i组数据,[每个训练点与测试点的距离+"@"+类标签列表]>
• Reducer：将输入数据（一组一组来）按照距离从小到大排序，统计前k个元素的不同类别出现的次数，并输出距离最近的前k个元素的分类结果。- 输入：<第i组数据,[每个训练点与测试点的距离+"@"+类标签列表]>- 按照距离从小到大排序，统计前k个元素的不同类别出现的次数，找出出现次数最多的类别。- 输出：<第i组数据,分类>

package KNN;importjava.io.BufferedReader;importjava.io.DataInput;importjava.io.DataOutput;importjava.io.File;importjava.io.FileReader;importjava.io.IOException;importjava.io.InputStreamReader;importjava.net.URI;importjava.nio.file.FileStore;importjava.nio.file.PathMatcher;importjava.nio.file.WatchService;importjava.nio.file.attribute.UserPrincipalLookupService;importjava.nio.file.spi.FileSystemProvider;importjava.util.ArrayList;importjava.util.List;importjava.util.Map;importjava.util.TreeMap;importjava.util.HashMap;importjava.util.Set;importjava.util.StringTokenizer;importjava.util.Iterator;importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.filecache.DistributedCache;importorg.apache.hadoop.fs.FileSystem;importorg.apache.hadoop.fs.Path;importorg.apache.hadoop.io.IntWritable;importorg.apache.hadoop.io.Text;importorg.apache.hadoop.io.WritableComparable;importorg.apache.hadoop.mapreduce.Job;importorg.apache.hadoop.mapreduce.Mapper;importorg.apache.hadoop.mapreduce.Reducer;importorg.apache.hadoop.mapreduce.lib.input.FileInputFormat;importorg.apache.hadoop.mapreduce.lib.output.FileOutputFormat;importorg.apache.hadoop.util.GenericOptionsParser;publicclass KNN 
{//Mapper:计算训练集中每个点与测试集中所有点的欧式距离，并将距离和对应的类标签输出。publicstaticclassTokenizerMapperextendsMapper<Object,Text,IntWritable,Text>{//List<String>类型的test变量,存储测试集staticList<String> test =newArrayList<String>();@Override//该函数在Mapper类运行前调用，用于初始化一些参数，读取缓存文件。protectedvoidsetup(Context context)throwsIOException,InterruptedException{//获取缓存文件的路径数组，这里只有一个缓存文件Path[] paths =DistributedCache.getLocalCacheFiles(context.getConfiguration());System.out.println(paths);//通过BufferedReader类读取缓存文件中的内容BufferedReader sb =newBufferedReader(newFileReader(paths[0].toUri().getPath()));String temp =null;//将每行数据存储到test变量中。while((temp = sb.readLine())!=null){
                test.add(temp);}
            sb.close();System.out.println("+++++++"+ test);}//计算两个Double类型数组之间的欧式距离privatedoubledistance(Double[] a,Double[] b){double sum =0.0;for(int i =0; i < a.length; i++){//累加(a[i] - b[i])^2
                sum +=Math.pow(a[i]- b[i],2);}returnMath.sqrt(sum);//开根号}//将距离和对应的类标签输出publicvoidmap(Object key,Text value,Context context)throwsIOException,InterruptedException{//输入：<行偏移量,文件内容>,缓存的是test测试集，输入的是train训练集//将输入数据按照逗号分割成一个字符串数组train[]String train[]= value.toString().split(",");//最后一个元素是样本点所属的类别标签。String label = train[train.length -1];//将train[]中除了最后一个元素以外的每个元素转换成Double类型，并存储到train_point数组中Double[] train_point =newDouble[4];for(int i =0; i < train.length -1; i++){
                train_point[i]=Double.valueOf(train[i]);}//遍历存储测试集的变量testfor(int i =0; i < test.size(); i++){//将每个测试点按照逗号分割成一个字符串数组test_poit1[]String test_poit1[]= test.get(i).toString().split(",");//将test_poit1[]中的每个元素转换成Double类型，并存储到test_poit数组Double[] test_poit =newDouble[4];for(int j =0; j < test_poit1.length; j++){
                    test_poit[j]=Double.valueOf(test_poit1[j]);}//将test_poit和train_point作为参数调用distance()函数计算它们之间的欧式距离//输出:<第i组数据,每个训练点与测试点的距离+"@"+类标签>
                context.write(newIntWritable(i),newText(String.valueOf(distance(test_poit, train_point))+"@"+ label));}}}//Reducer:将输入数据按照距离从小到大排序，统计前k个元素的不同类别出现的次数，并输出距离最近的前k个元素的分类结果。publicstaticclassIntSumReducerextendsReducer<IntWritable,Text,IntWritable,Text>{//定义一个Text类型的变量result，用于存储分类结果。privateText result =newText();//定义变量用于设置k的值。int k;//setup()在Reducer开始处理数据之前初始化。protectedvoidsetup(Context context)throwsIOException,InterruptedException{//context是上下文对象，用于传递参数和获取配置信息。Configuration conf=context.getConfiguration();//获取名为"K"的配置参数的值，并将其转化为int类型的值，如果未找到该配置参数，则默认值为1。
            k=conf.getInt("K",1);}//将输入数据按照距离从小到大排序，统计前k个元素的不同类别出现的次数，并输出距离最近的前k个元素的分类结果。publicvoidreduce(IntWritable key,Iterable<Text> values,Context context)throwsIOException,InterruptedException{//输入：<第i组数据,[每个训练点与测试点的距离+"@"+类标签列表]>//定义一个TreeMap，用于将输入数据按照距离从小到大排序，其中键为距离，值为该距离所对应的类标签列表。TreeMap<Double,String> treemap =newTreeMap<Double,String>();for(Text val: values){//将输入数据按照"@"进行分割，分别存储距离和类标签列表。String distance_lable[]= val.toString().split("@");for(int i =0; i < distance_lable.length -1; i = i +2){//遍历类标签列表，每次i+2,跳过类标签//将输入数据按照距离从小到大排序,将当前距离和对应的类标签列表存储到TreeMap中。
                    treemap.put(Double.valueOf(distance_lable[i]), distance_lable[i +1]);}}//获取TreeMap的键集合，并返回一个迭代器，用于遍历TreeMap中所有的键。Iterator<Double> it = treemap.keySet().iterator();//定义一个HashMap，用于统计前k个元素的不同类别出现的次数。Map<String,Integer> map =newHashMap<String,Integer>();int num =0;//统计前k个元素的不同类别出现的次数while(it.hasNext()){//获取 TreeMap 的下一个键Double key1 = it.next();//如果 Map 中已经存在当前键所代表的值，则表示该值已经出现过，需要将对应键的值加 1if(map.containsKey(treemap.get(key1))){//获取当前键所对应的值在 Map 中对应的计数int temp = map.get(treemap.get(key1));//将计数加 1，然后更新 Map 对应键所代表的值
                    map.put(treemap.get(key1), temp +1);}else//如果 Map 中不存在当前键所代表的值，则表示该值还没有出现过，需要在 Map 中添加一个新的元素{//将当前键所代表的值加入 Map，并设定计数为 1
                    map.put(treemap.get(key1),1);}//记录已经处理过的元素个数
                num++;if(num > k)break;}//找出前k个数据里出现次数最多的类别Iterator<String> it1 = map.keySet().iterator();String label = it1.next();//并将其对应的值存储在`count`变量中int count = map.get(label);while(it1.hasNext()){//当还有键未被遍历时，执行循环String now = it1.next();if(count < map.get(now)){//更新出现次数最多的类别
                    label = now;
                    count = map.get(label);}}//循环直到所有键都遍历完毕
            result.set(label);//输出：<第i组数据,分类>
            context.write(key, result);}}publicstaticvoidmain(String[] args)throwsException{Configuration conf =newConfiguration();String[] otherArgs =newString[]{"hdfs://localhost:9000/user/hadoop/input/train.csv","hdfs://localhost:9000/user/hadoop/output"};if(otherArgs.length <2){System.err.println("Usage: wordcount <in> [<in>...] <out>");System.exit(2);}
        conf.setInt("K",5);Job job1 =Job.getInstance(conf,"KNN");
        job1.setJarByClass(KNN.class);//设置分布式缓存文件
        job1.addCacheFile(newURI("hdfs://localhost:9000/user/hadoop/input/test.csv"));
        job1.setMapperClass(TokenizerMapper.class);
        job1.setReducerClass(IntSumReducer.class);
        job1.setMapOutputKeyClass(IntWritable.class);
        job1.setMapOutputValueClass(Text.class);
        job1.setOutputKeyClass(IntWritable.class);
        job1.setOutputValueClass(Text.class);for(int i =0; i < otherArgs.length -1; i++){FileInputFormat.addInputPath(job1,newPath(otherArgs[i]));}FileOutputFormat.setOutputPath(job1,newPath(otherArgs[otherArgs.length -1]));System.exit(job1.waitForCompletion(true)?0:1);}}

🥕在集群上提交作业并执行

• 修改缓存路径、输入、输出路径（这里的路径其实不固定，以下是个人使用路径，与终端操作对应）- 输入：hdfs://10.102.0.198:9000/user/bigdata_学号/input/train.csv- 输出：hdfs://10.102.0.198:9000/user/bigdata_学号/output- 缓存路径：hdfs://10.102.0.198:9000/user/bigdata_学号/input/test.csv
• 导出export包（这里的具体操作不清楚的可看实验二）
• 将数据包及导出的jar包上传至集群，终端操作依次如下：scp test.csv bigdata_学号@10.102.0.198:/home/bigdata_学号scp train.csv bigdata_学号@10.102.0.198:/home/bigdata_学号scp KNN.jar bigdata_学号@10.102.0.198:/home/bigdata_学号
• 登录ssh：ssh bigdata_学号@10.102.0.198
• 建好input文件夹，导入相关文件，并删除原有的output，终端操作依次如下：hdfs dfs -mkdir inputhdfs dfs -put train.csv /user/bigdata_学号/input/hdfs dfs -put test.csv /user/bigdata_学号/input/hadoop fs -rm-r output
• 运行：hadoop jar KNN.jar> 最后可在集群查看提交情况>

🐇非并行化

采用python（knn函数自带），并在Anaconda的JupyterLab运行。

import numpy as np
import pandas as pd
from sklearn.neighbors import KNeighborsClassifier 
import matplotlib as mpl
import matplotlib.pyplot as plt

# header参数指定标题的行，默认为0.如果没有标题，则使用None
data = pd.read_csv('iris_train1.csv',header=0)
data2=pd.read_csv('iris_test1.csv',header=0)print(data.head())print(data2.head())# 将文本数据映射成数值类型
data['Species']=data['Species'].map({"versicolor":0,"setosa":1,"virginica":2})
data2['Species']=data2['Species'].map({"versicolor":0,"setosa":1,"virginica":2})# print(data.duplicated().any())
data.drop_duplicates(inplace=True)print(data.shape)print(data2.shape)

# 提取出每个类中鸢尾花数据
t0=data[data['Species']==0]
t1=data[data['Species']==1]
t2=data[data['Species']==2]

# 打乱每个类别的数据
t0=t0.sample(len(t0),random_state=0)
t1=t1.sample(len(t1),random_state=0)
t2=t2.sample(len(t2),random_state=0)

# 分配训练集和测试集
train_X=pd.concat([t0.iloc[:30,:-1],t1.iloc[:30,:-1],t2.iloc[:30,:-1]],axis=0)
train_y=pd.concat([t0.iloc[:30,-1],t1.iloc[:30,-1],t2.iloc[:30,-1]],axis=0)

df=pd.DataFrame(data2)
test_X=df[["Sepal.Length","Sepal.Width","Petal.Length","Petal.Width"]]
test_y=df['Species']

# 训练与测试
knn=KNeighborsClassifier(3)
knn.fit(train_X,train_y)
result=knn.predict(test_X)print(result)
final =[]for i in result:if i ==0:
        final.append("versicolor")elif i ==1:
        final.append("setosa")else:
        final.append("virginica")print(final)

# 可视化# 设置画布大小
plt.figure(figsize=(20,10))# 设置字体为黑体，支持中文显示
mpl.rcParams['font.family']='SimHei'# 设置中文字体，可以正常显示负号
mpl.rcParams['axes.unicode_minus']=False# 绘制训练集数据
plt.subplot(121)
plt.scatter(x=t0['Sepal.Length'][:40],y=t0['Petal.Length'][:40],color='r',label='versicolor')
plt.scatter(x=t1['Sepal.Length'][:40],y=t1['Petal.Length'][:40],color='g',label='setosa')
plt.scatter(x=t2['Sepal.Length'][:40],y=t2['Petal.Length'][:40],color='b',label='virginica')
plt.xlabel('花萼长度')
plt.ylabel('花瓣长度')
plt.legend(loc='best')

plt.subplot(122)
right = test_X[result == test_y]
wrong = test_X[result != test_y]
plt.scatter(x=right['Sepal.Length'], y=right['Petal.Length'], color='c', label="right", marker="x")
plt.scatter(x=wrong['Sepal.Length'], y=wrong['Petal.Length'], color='m', label="wrong", marker=">")
plt.xlabel('花萼长度')
plt.ylabel('花瓣长度')
plt.legend(loc='best')
plt.show()

Anaconda是在之前上数据导论的时候安装的（好像当时还挺难装 ，具体安装教程之后再补吧（当时没记可恶 。

以下是完整python代码（没装Anaconda问题不大，pycharm或者vs都可以

import numpy as np
import pandas as pd
from sklearn.neighbors import KNeighborsClassifier
import matplotlib as mpl
import matplotlib.pyplot as plt

# header参数指定标题的行，默认为0.如果没有标题，则使用None
data = pd.read_csv('iris_train1.csv',header=0)
data2=pd.read_csv('iris_test1.csv',header=0)print(data.head())print(data2.head())# 将文本数据映射成数值类型
data['Species']=data['Species'].map({"versicolor":0,"setosa":1,"virginica":2})
data2['Species']=data2['Species'].map({"versicolor":0,"setosa":1,"virginica":2})# print(data.duplicated().any())
data.drop_duplicates(inplace=True)print(data.shape)print(data2.shape)# 提取出每个类中鸢尾花数据
t0=data[data['Species']==0]
t1=data[data['Species']==1]
t2=data[data['Species']==2]# 打乱每个类别的数据
t0=t0.sample(len(t0),random_state=0)
t1=t1.sample(len(t1),random_state=0)
t2=t2.sample(len(t2),random_state=0)# 分配训练集和测试集
train_X=pd.concat([t0.iloc[:30,:-1],t1.iloc[:30,:-1],t2.iloc[:30,:-1]],axis=0)
train_y=pd.concat([t0.iloc[:30,-1],t1.iloc[:30,-1],t2.iloc[:30,-1]],axis=0)

df=pd.DataFrame(data2)
test_X=df[["Sepal.Length","Sepal.Width","Petal.Length","Petal.Width"]]
test_y=df['Species']# 训练与测试
knn=KNeighborsClassifier(3)
knn.fit(train_X,train_y)
result=knn.predict(test_X)print(result)
final =[]for i in result:if i ==0:
        final.append("versicolor")elif i ==1:
        final.append("setosa")else:
        final.append("virginica")print(final)# 可视化# 设置画布大小
plt.figure(figsize=(20,10))# 设置字体为黑体，支持中文显示
mpl.rcParams['font.family']='SimHei'# 设置中文字体，可以正常显示负号
mpl.rcParams['axes.unicode_minus']=False# 绘制训练集数据
plt.subplot(121)
plt.scatter(x=t0['Sepal.Length'][:40],y=t0['Petal.Length'][:40],color='r',label='versicolor')
plt.scatter(x=t1['Sepal.Length'][:40],y=t1['Petal.Length'][:40],color='g',label='setosa')
plt.scatter(x=t2['Sepal.Length'][:40],y=t2['Petal.Length'][:40],color='b',label='virginica')
plt.xlabel('花萼长度')
plt.ylabel('花瓣长度')
plt.legend(loc='best')

plt.subplot(122)
right = test_X[result == test_y]
wrong = test_X[result != test_y]
plt.scatter(x=right['Sepal.Length'], y=right['Petal.Length'], color='c', label="right", marker="x")
plt.scatter(x=wrong['Sepal.Length'], y=wrong['Petal.Length'], color='m', label="wrong", marker=">")
plt.xlabel('花萼长度')
plt.ylabel('花瓣长度')
plt.legend(loc='best')
plt.show()