使用MapReduce实现knn算法

2024-06-20 18:18

本文主要是介绍使用MapReduce实现knn算法,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

算法的流程

(1)首先将训练集以共享文件的方式分发到各个map节点

(2)每一个map节点主要<LongWritable ,Text,,LongWritable,ListWritable<DoubleWritable>> LongWritable 主要就是文件的偏移地址,保证唯一。ListWritable主要就是最近的类别。

Reduce节点主要计算出,每一个要预测节点的类别。

package knn;


public class Distance {

public static double EuclideanDistance(double[] a, double[] b)
throws Exception {
if (a.length != b.length)
throw new Exception("size not compatible!");
double sum = 0.0;
for (int i = 0; i < a.length; i++) {
sum += Math.pow(a[i] - b[i], 2);
}
return Math.sqrt(sum);
}
}

package knn;


import java.io.BufferedReader;


/**
 * KNearestNeigbour Classifier each instance in training set is of form
 * a1,a2,a3...an,l1 in which l1 represents the label. and each instance in
 * predict set is of form a1,a2,a3...an,-1,in which -1 is the label we want to
 * specify. In my algorithm,I assume that the trainning set is relatively small
 * so we can load them in memory and the predict set is large another thing we
 * need to pay attention to is that all our test instances are all in one file
 * so that the index of line is unique to each instance.
 * 
 */
public class KNearestNeighbour {
public static class KNNMap
extends
Mapper<LongWritable, Text, LongWritable, ListWritable<DoubleWritable>> {
private int k;
private ArrayList<Instance> trainSet;


@Override
protected void setup(Context context) throws IOException,
InterruptedException {
k = context.getConfiguration().getInt("k", 1);
trainSet = new ArrayList<Instance>();


Path[] trainFile = DistributedCache.getLocalCacheFiles(context
.getConfiguration());
// add all the tranning instances into attributes
BufferedReader br = null;
String line;
for (int i = 0; i < trainFile.length; i++) {
br = new BufferedReader(new FileReader(trainFile[0].toString()));
while ((line = br.readLine()) != null) {
Instance trainInstance = new Instance(line);
System.out.println(trainInstance.toString());
trainSet.add(trainInstance);
}
}
}


/**
* find the nearest k labels and put them in an object of type
* ListWritable. and emit <textIndex,lableList>
*/
@Override
public void map(LongWritable textIndex, Text textLine, Context context)
throws IOException, InterruptedException {
System.out.println(textLine.toString());
// distance stores all the current nearst distance value
// . trainLable store the corresponding lable
ArrayList<Double> distance = new ArrayList<Double>(k);
ArrayList<DoubleWritable> trainLable = new ArrayList<DoubleWritable>(
k);
for (int i = 0; i < k; i++) {
distance.add(Double.MAX_VALUE);
trainLable.add(new DoubleWritable(-1.0));
}
ListWritable<DoubleWritable> lables = new ListWritable<DoubleWritable>(
DoubleWritable.class);
Instance testInstance = new Instance(textLine.toString());
for (int i = 0; i < trainSet.size(); i++) {
try {
double dis = Distance.EuclideanDistance(trainSet.get(i)
.getAtrributeValue(), testInstance
.getAtrributeValue());
int index = indexOfMax(distance);
if (dis < distance.get(index)) {
distance.remove(index);
trainLable.remove(index);
distance.add(dis);
trainLable.add(new DoubleWritable(trainSet.get(i)
.getLable()));
}
} catch (Exception e) {
e.printStackTrace();
}
}
lables.setList(trainLable);
context.write(textIndex, lables);
}


/**
* return the index of the maximum number of an array

* @param array
* @return
*/
public int indexOfMax(ArrayList<Double> array) {
int index = -1;
Double min = Double.MIN_VALUE;
for (int i = 0; i < array.size(); i++) {
if (array.get(i) > min) {
min = array.get(i);
index = i;
}
}
return index;
}
}


public static class KNNReduce
extends
Reducer<LongWritable, ListWritable<DoubleWritable>, NullWritable, DoubleWritable> {


@Override
public void reduce(LongWritable index,
Iterable<ListWritable<DoubleWritable>> kLables, Context context)
throws IOException, InterruptedException {
/**
* each index can actually have one list because of the assumption
* that the particular line index is unique to one instance.
*/
DoubleWritable predictedLable = new DoubleWritable();
for (ListWritable<DoubleWritable> val : kLables) {
try {
predictedLable = valueOfMostFrequent(val);
break;
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
context.write(NullWritable.get(), predictedLable);
}


public DoubleWritable valueOfMostFrequent(
ListWritable<DoubleWritable> list) throws Exception {
if (list.isEmpty())
throw new Exception("list is empty!");
else {
HashMap<DoubleWritable, Integer> tmp = new HashMap<DoubleWritable, Integer>();
for (int i = 0; i < list.size(); i++) {
if (tmp.containsKey(list.get(i))) {
Integer frequence = tmp.get(list.get(i)) + 1;
tmp.remove(list.get(i));
tmp.put(list.get(i), frequence);
} else {
tmp.put(list.get(i), new Integer(1));
}
}
// find the value with the maximum frequence.
DoubleWritable value = new DoubleWritable();
Integer frequence = new Integer(Integer.MIN_VALUE);
Iterator<Entry<DoubleWritable, Integer>> iter = tmp.entrySet()
.iterator();
while (iter.hasNext()) {
Map.Entry<DoubleWritable, Integer> entry = (Map.Entry<DoubleWritable, Integer>) iter
.next();
if (entry.getValue() > frequence) {
frequence = entry.getValue();
value = entry.getKey();
}
}
return value;
}
}
}


public static void main(String[] args) throws IOException,
InterruptedException, ClassNotFoundException {
Job kNNJob = new Job();
kNNJob.setJobName("kNNJob");
kNNJob.setJarByClass(KNearestNeighbour.class);
DistributedCache.addCacheFile(URI.create(args[2]), kNNJob
.getConfiguration());
kNNJob.getConfiguration().setInt("k", Integer.parseInt(args[3]));


kNNJob.setMapperClass(KNNMap.class);
kNNJob.setMapOutputKeyClass(LongWritable.class);
kNNJob.setMapOutputValueClass(ListWritable.class);


kNNJob.setReducerClass(KNNReduce.class);
kNNJob.setOutputKeyClass(NullWritable.class);
kNNJob.setOutputValueClass(DoubleWritable.class);


kNNJob.setInputFormatClass(TextInputFormat.class);
kNNJob.setOutputFormatClass(TextOutputFormat.class);
FileInputFormat.addInputPath(kNNJob, new Path(args[0]));
FileOutputFormat.setOutputPath(kNNJob, new Path(args[1]));


kNNJob.waitForCompletion(true);
System.out.println("finished!");
}
}

package knn;


public class Instance {
private double[] attributeValue;
private double lable;


/**
* a line of form a1 a2 ...an lable

* @param line
*/
public Instance(String line) {
System.out.println(line);
String[] value = line.split(" ");
attributeValue = new double[value.length - 1];
for (int i = 0; i < attributeValue.length; i++) {
attributeValue[i] = Double.parseDouble(value[i]);
System.out.print(attributeValue[i] + "\t");
}
lable = Double.parseDouble(value[value.length - 1]);
System.out.println(lable);
}


public double[] getAtrributeValue() {
return attributeValue;
}


public double getLable() {
return lable;
}
}

package knn;


import java.io.DataInput;


public class ListWritable<T extends Writable> implements Writable {
private List<T> list;
private Class<T> clazz;


public ListWritable() {
list = null;
clazz = null;
}


public ListWritable(Class<T> clazz) {
this.clazz = clazz;
list = new ArrayList<T>();
}


public void setList(List<T> list) {
this.list = list;
}


public boolean isEmpty() {
return list.isEmpty();
}


public int size() {
return list.size();
}


public void add(T element) {
list.add(element);
}


public void add(int index, T element) {
list.add(index, element);
}


public T get(int index) {
return list.get(index);
}


public T remove(int index) {
return list.remove(index);
}


public void set(int index, T element) {
list.set(index, element);
}


@Override
public void write(DataOutput out) throws IOException {
out.writeUTF(clazz.getName());
out.writeInt(list.size());
for (T element : list) {
element.write(out);
}
}


@SuppressWarnings("unchecked")
@Override
public void readFields(DataInput in) throws IOException {
try {
clazz = (Class<T>) Class.forName(in.readUTF());
} catch (ClassNotFoundException e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
}
int count = in.readInt();
this.list = new ArrayList<T>();
for (int i = 0; i < count; i++) {
try {
T obj = clazz.newInstance();
obj.readFields(in);
list.add(obj);
} catch (InstantiationException e) {
e.printStackTrace();
} catch (IllegalAccessException e) {
e.printStackTrace();
}
}
}


}


训练集

1.0 2.0 3.0 1
1.0 2.1 3.1 1
0.9 2.2 2.9 1
3.4 6.7 8.9 2
3.0 7.0 8.7 2
3.3 6.9 8.8 2
2.5 3.3 10.0 3
2.4 2.9 8.0 3

这篇关于使用MapReduce实现knn算法的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1078890

相关文章

基于Python实现一个图片拆分工具

《基于Python实现一个图片拆分工具》这篇文章主要为大家详细介绍了如何基于Python实现一个图片拆分工具,可以根据需要的行数和列数进行拆分,感兴趣的小伙伴可以跟随小编一起学习一下... 简单介绍先自己选择输入的图片,默认是输出到项目文件夹中,可以自己选择其他的文件夹,选择需要拆分的行数和列数,可以通过

Python中将嵌套列表扁平化的多种实现方法

《Python中将嵌套列表扁平化的多种实现方法》在Python编程中,我们常常会遇到需要将嵌套列表(即列表中包含列表)转换为一个一维的扁平列表的需求,本文将给大家介绍了多种实现这一目标的方法,需要的朋... 目录python中将嵌套列表扁平化的方法技术背景实现步骤1. 使用嵌套列表推导式2. 使用itert

使用Docker构建Python Flask程序的详细教程

《使用Docker构建PythonFlask程序的详细教程》在当今的软件开发领域,容器化技术正变得越来越流行,而Docker无疑是其中的佼佼者,本文我们就来聊聊如何使用Docker构建一个简单的Py... 目录引言一、准备工作二、创建 Flask 应用程序三、创建 dockerfile四、构建 Docker

Python使用vllm处理多模态数据的预处理技巧

《Python使用vllm处理多模态数据的预处理技巧》本文深入探讨了在Python环境下使用vLLM处理多模态数据的预处理技巧,我们将从基础概念出发,详细讲解文本、图像、音频等多模态数据的预处理方法,... 目录1. 背景介绍1.1 目的和范围1.2 预期读者1.3 文档结构概述1.4 术语表1.4.1 核

Python使用pip工具实现包自动更新的多种方法

《Python使用pip工具实现包自动更新的多种方法》本文深入探讨了使用Python的pip工具实现包自动更新的各种方法和技术,我们将从基础概念开始,逐步介绍手动更新方法、自动化脚本编写、结合CI/C... 目录1. 背景介绍1.1 目的和范围1.2 预期读者1.3 文档结构概述1.4 术语表1.4.1 核

在Linux中改变echo输出颜色的实现方法

《在Linux中改变echo输出颜色的实现方法》在Linux系统的命令行环境下,为了使输出信息更加清晰、突出,便于用户快速识别和区分不同类型的信息,常常需要改变echo命令的输出颜色,所以本文给大家介... 目python录在linux中改变echo输出颜色的方法技术背景实现步骤使用ANSI转义码使用tpu

Conda与Python venv虚拟环境的区别与使用方法详解

《Conda与Pythonvenv虚拟环境的区别与使用方法详解》随着Python社区的成长,虚拟环境的概念和技术也在不断发展,:本文主要介绍Conda与Pythonvenv虚拟环境的区别与使用... 目录前言一、Conda 与 python venv 的核心区别1. Conda 的特点2. Python v

Spring Boot中WebSocket常用使用方法详解

《SpringBoot中WebSocket常用使用方法详解》本文从WebSocket的基础概念出发,详细介绍了SpringBoot集成WebSocket的步骤,并重点讲解了常用的使用方法,包括简单消... 目录一、WebSocket基础概念1.1 什么是WebSocket1.2 WebSocket与HTTP

C#中Guid类使用小结

《C#中Guid类使用小结》本文主要介绍了C#中Guid类用于生成和操作128位的唯一标识符,用于数据库主键及分布式系统,支持通过NewGuid、Parse等方法生成,感兴趣的可以了解一下... 目录前言一、什么是 Guid二、生成 Guid1. 使用 Guid.NewGuid() 方法2. 从字符串创建

Python使用python-can实现合并BLF文件

《Python使用python-can实现合并BLF文件》python-can库是Python生态中专注于CAN总线通信与数据处理的强大工具,本文将使用python-can为BLF文件合并提供高效灵活... 目录一、python-can 库:CAN 数据处理的利器二、BLF 文件合并核心代码解析1. 基础合