使用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

相关文章

postgresql使用UUID函数的方法

《postgresql使用UUID函数的方法》本文给大家介绍postgresql使用UUID函数的方法,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友参考下吧... 目录PostgreSQL有两种生成uuid的方法。可以先通过sql查看是否已安装扩展函数,和可以安装的扩展函数

Python实现终端清屏的几种方式详解

《Python实现终端清屏的几种方式详解》在使用Python进行终端交互式编程时,我们经常需要清空当前终端屏幕的内容,本文为大家整理了几种常见的实现方法,有需要的小伙伴可以参考下... 目录方法一:使用 `os` 模块调用系统命令方法二:使用 `subprocess` 模块执行命令方法三:打印多个换行符模拟

SpringBoot+EasyPOI轻松实现Excel和Word导出PDF

《SpringBoot+EasyPOI轻松实现Excel和Word导出PDF》在企业级开发中,将Excel和Word文档导出为PDF是常见需求,本文将结合​​EasyPOI和​​Aspose系列工具实... 目录一、环境准备与依赖配置1.1 方案选型1.2 依赖配置(商业库方案)二、Excel 导出 PDF

Python实现MQTT通信的示例代码

《Python实现MQTT通信的示例代码》本文主要介绍了Python实现MQTT通信的示例代码,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一... 目录1. 安装paho-mqtt库‌2. 搭建MQTT代理服务器(Broker)‌‌3. pytho

如何使用Lombok进行spring 注入

《如何使用Lombok进行spring注入》本文介绍如何用Lombok简化Spring注入,推荐优先使用setter注入,通过注解自动生成getter/setter及构造器,减少冗余代码,提升开发效... Lombok为了开发环境简化代码,好处不用多说。spring 注入方式为2种,构造器注入和setter

MySQL中比较运算符的具体使用

《MySQL中比较运算符的具体使用》本文介绍了SQL中常用的符号类型和非符号类型运算符,符号类型运算符包括等于(=)、安全等于(=)、不等于(/!=)、大小比较(,=,,=)等,感兴趣的可以了解一下... 目录符号类型运算符1. 等于运算符=2. 安全等于运算符<=>3. 不等于运算符<>或!=4. 小于运

使用zip4j实现Java中的ZIP文件加密压缩的操作方法

《使用zip4j实现Java中的ZIP文件加密压缩的操作方法》本文介绍如何通过Maven集成zip4j1.3.2库创建带密码保护的ZIP文件,涵盖依赖配置、代码示例及加密原理,确保数据安全性,感兴趣的... 目录1. zip4j库介绍和版本1.1 zip4j库概述1.2 zip4j的版本演变1.3 zip4

Python 字典 (Dictionary)使用详解

《Python字典(Dictionary)使用详解》字典是python中最重要,最常用的数据结构之一,它提供了高效的键值对存储和查找能力,:本文主要介绍Python字典(Dictionary)... 目录字典1.基本特性2.创建字典3.访问元素4.修改字典5.删除元素6.字典遍历7.字典的高级特性默认字典

使用Python构建一个高效的日志处理系统

《使用Python构建一个高效的日志处理系统》这篇文章主要为大家详细讲解了如何使用Python开发一个专业的日志分析工具,能够自动化处理、分析和可视化各类日志文件,大幅提升运维效率,需要的可以了解下... 目录环境准备工具功能概述完整代码实现代码深度解析1. 类设计与初始化2. 日志解析核心逻辑3. 文件处

python生成随机唯一id的几种实现方法

《python生成随机唯一id的几种实现方法》在Python中生成随机唯一ID有多种方法,根据不同的需求场景可以选择最适合的方案,文中通过示例代码介绍的非常详细,需要的朋友们下面随着小编来一起学习学习... 目录方法 1:使用 UUID 模块(推荐)方法 2:使用 Secrets 模块(安全敏感场景)方法