首页 > 代码库 > MapReduce实现线性回归
MapReduce实现线性回归
1. 软件版本:
Hadoop2.6.0(IDEA中源码编译使用CDH5.7.3,对应Hadoop2.6.0),集群使用原生Hadoop2.6.4,JDK1.8,Intellij IDEA 14 。源码可以在https://github.com/fansy1990/linear_regression 下载。
2. 实现思路:
本博客实现的是一元一次线性方程,等于是最简单的线性方程了,采用的是Couresa里面的机器学习中的大数据线性方程的方法来更新参数值的(即随机梯度下降方法,当然也可以使用批量梯度下降方法来实现,只是在LinearRegressionJob中实现的不一样而已),如果对随机梯度下降或者批量梯度下降不了解的话,需要先去看看。下面是实现思路:
2.1 Shuffle Data(打乱数据):
如果要采用随机梯度下降的话,那么需要保持原始数据随机,所以这里的第一步就是随机打乱原始数据。采用的思路是:在Mapper端输出随机值作为key,输出当前记录作为value,在Reducer端直接遍历每个key的所有values,直接输出value以及NullWritable.get即可。
在这里添加一个额外的参数randN,这个参数表示在Mapper端随机值时,多少个原始数据使用同一个随机值,如果randN为1,那么每个原始数据都会使用一个随机值作为key,如果randN为2,那么每两个原始数据使用一个随机值,如果randN为0或小于0,那么所有数据都使用同一个随机值(注意,这个时候其实在Reducer端的values其实也是乱序的,请读者思考为什么?)。
其Mapper中map核心实现如下所示
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { if(randN <= 0) { // 如果randN 比0小,那么不再次打乱数据 context.write(randFloatKey,value); return ; } if(++countI >= randN){// 如果randN等于1,那么每次随机的值都是不一样的 randFloatKey.set(random.nextFloat()); countI =0; } context.write(randFloatKey,value); }
2.2 Linear Regression(线性回归):
线性回归采用随机梯度下降的方法来更新theta0和theta1 (只实现了一元一次,所以只有两个参数),每个Mapper都会使用同样的初始化参数(theta0=1和theta1=0),在每个Mapper中使用自己的数据来更新theta0和theta1,更新的公式为:
theta0 = theta0 -alpha*(h(x)-y)x theta1 = theta1 -alpha*(h(x)-y)x其中,h(x)= theta0 + theta1 * x ;同时,需要注意这里的更新是同步更新,其核心代码如下所示:
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { float[] xy = Utils.str2float(value.toString().split(splitter)); float x = xy[0]; float y = xy[1]; // 同步更新 theta0 and theta1 lastTheta0 = theta0; theta0 -= alpha *(theta0+theta1* x - y) * x; // 保持theta0 和theta1 不变 theta1 -= alpha *(lastTheta0 + theta1 * x -y) * x;// 保持theta0 和theta1 不变 }
然后在每个Mapper的cleanup函数中直接输出theta的参数值即可
protected void cleanup(Context context) throws IOException, InterruptedException { theta0_1.set(theta0 + splitter + theta1); context.write(theta0_1,NullWritable.get()); }
由于在每个mapper中已经更新了theta的各个参数值,所以不需要使用reducer即可;同时,由于测试数据比较小,所以设置mapreduce.input.fileinputformat.split.maxsize的大小,读者需要根据自己实际数据的大小来设置,其Driver类核心代码如下所示:
conf.setLong("mapreduce.input.fileinputformat.split.maxsize",700L);// 获取多个mapper; job.setNumReduceTasks(0);
2.3 Combine Theta (合并参数值):
在2.2步中已经算得了各个theta值,那么应该如何来合并这些求得得各个theta值呢?可以直接用平均值么?对于一元一次线性回归是可以直接使用平均值来作为最终合并后的theta值的,但是针对其他的线性回归(特指有多个局部最小值的线性回归,这样求得的多个theta值合并就会有问题了)。
如果只是使用平均值的话,那么在2.2步其实加一个Reducer就可以完成了,这里提出了一种另外的方式来合并theta值,即采用各个theta值的全局误差作为参数来进行加权。所以,在Mapper的setup中会读取2.2中的多个输出theta值,在map函数中针对各个原始数据求其误差,输出到reducer的数据为theta值和其误差;其核心代码如下所示:
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { float[] xy = Utils.str2float(value.toString().split(splitter)); for(int i =0;i<thetas.size() ;i++){ // error = (theta0 + theta1 * x - y) ^2 thetaErrors[i] += (thetas.get(i)[0]+ thetas.get(i)[1] * xy[0] -xy[1]) * (thetas.get(i)[0]+ thetas.get(i)[1] * xy[0] -xy[1]) ; thetaNumbers[i]+= 1; } }
protected void cleanup(Context context) throws IOException, InterruptedException { for(int i =0;i<thetas.size() ;i++){ theta.set(thetas.get(i)); floatAndLong.set(thetaErrors[i],thetaNumbers[i]); context.write(theta,floatAndLong); } }
在Reducer端,直接针对每个键(也就是theta值)把各个误差加起来,在cleanup函数中采用加权来合并theta值,其核心代码如下所示:
protected void reduce(FloatAndFloat key, Iterable<FloatAndLong> values, Context context) throws IOException, InterruptedException { float sumF = 0.0f; long sumL = 0L ; for(FloatAndLong value:values){ sumF +=value.getSumFloat(); sumL += value.getSumLong(); } theta_error.add(new float[]{key.getTheta0(),key.getTheta1(), (float)Math.sqrt((double)sumF / sumL)}); logger.info("theta:{}, error:{}", new Object[]{key.toString(),Math.sqrt(sumF/sumL)}); }
protected void cleanup(Context context) throws IOException, InterruptedException { // 如何加权? // 方式1:如果误差越小,那么说明权重应该越大; // 方式2:直接平均值 float [] theta_all = new float[2]; if("average".equals(method)){ // theta_all = theta_error.get(0); for(int i=0;i< theta_error.size();i++){ theta_all[0] += theta_error.get(i)[0]; theta_all[1] += theta_error.get(i)[1]; } theta_all[0] /= theta_error.size(); theta_all[1] /= theta_error.size(); } else { float sumErrors = 0.0f; for(float[] d:theta_error){ sumErrors += 1/d[2]; } for(float[] d: theta_error){ theta_all[0] += d[0] * 1/d[2] /sumErrors; theta_all[1] += d[1] * 1/d[2] /sumErrors; } } context.write(new FloatAndFloat(theta_all),NullWritable.get()); }
2.4 验证
这里的验证指的是使用2.3步求的得合并后的theta值求全局误差,由于在2.3步也求得了各个theta值的全局误差,所以这里可以对比看下哪个theta值最优;其Mapper可以直接使用2.3步骤的mapper,而reducer也类似2.3步骤中的reducer,只是最终输出就不需要cleanup中的合并了。
3. 运行结果:
3.1 shuffle Job
测试类:
public static void main(String[] args) throws Exception { args = new String[]{ "hdfs://master:8020/user/fanzhe/linear_regression.txt", "hdfs://master:8020/user/fanzhe/shuffle_out", "1" } ; ToolRunner.run(Utils.getConf(),new ShuffleDataJob(),args); }
原始数据:(可以在源码中的resource目录中下载 linear_regression.txt)
6.1101,17.592 5.5277,9.1302 8.5186,13.662 。。。Shuffle输出:
每次输出应该都是不一样的(使用了随机数),可以看到数据确实被随机化了。
3.2 Linear Regression
测试类:
public static void main(String[] args) throws Exception { // <input> <output> <theta0;theta1;alpha> <splitter> // 注意第三个参数使用分号分割 args = new String[]{ "hdfs://master:8020/user/fanzhe/shuffle_out", "hdfs://master:8020/user/fanzhe/linear_regression", "1;0;0.01", "," } ; ToolRunner.run(Utils.getConf(),new LinearRegressionJob(),args); }查看输出结果:
从输出结果可以看出,两个结果相差还是很大的,这个主要是因为测试数据比较少的原因,如果数据比较大,并且被很好的shuffle的话,那么这两个值应该是相差不大的;
3.3 Combine Theta
测试类:
public static void main(String[] args) throws Exception { // <input> <output> <theta_path> <splitter> <average|weight> args = new String[]{ "hdfs://master:8020/user/fanzhe/shuffle_out", "hdfs://master:8020/user/fanzhe/single_linear_regression_error", "hdfs://master:8020/user/fanzhe/linear_regression", ",", "weight" } ; ToolRunner.run(Utils.getConf(),new SingleLinearRegressionError(),args); }这里设置的合并theta值的方式使用加权,读者可以设置为average,从而使用平均值;
结果:
根据日志可以看出theta参数值选取下面的一个,其误差会比较小,合并后的参数值为:
看到其结果是在两个theta参数值之间。
如果是平均值,那么其输出结果为:
3.4 验证
验证测试类:
public static void main(String[] args) throws Exception { // <input> <output> <theta_path> <splitter> args = new String[]{ "hdfs://master:8020/user/fanzhe/shuffle_out", "hdfs://master:8020/user/fanzhe/last_linear_regression_error", "hdfs://master:8020/user/fanzhe/single_linear_regression_error", ",", } ; ToolRunner.run(Utils.getConf(),new LastLinearRegressionError(),args); }输出结果为:
从结果中可以看出,合并后的结果并没有原来其中的一个Theta参数组值的效果好,不过这个也可能和数据量有关,根据输出结果,也可以把合并后的theta值以及合并前的对比,然后使用最优的theta来作为最后的输出。
如果是平均值,那么其输出结果为:
从上面的结果可以看到加权的组合比平均值的组合效果好点;
4. 总结
1. 改算法只针对有一个局部最优解(也就是全局最优解)的情况,否则,在合并阶段会有问题;
2. 通过小量数据验证,使用合并后的效果并没有使用合并前的最优解的效果好,这个可能是数据问题,待验证;
3. 通过很直观的想象,一般情况下使用加权组合要比平均组好效果好;
分享,成长,快乐
转载请注明blog地址:http://blog.csdn.net/fansy1990
MapReduce实现线性回归
声明:以上内容来自用户投稿及互联网公开渠道收集整理发布,本网站不拥有所有权,未作人工编辑处理,也不承担相关法律责任,若内容有误或涉及侵权可进行投诉: 投诉/举报 工作人员会在5个工作日内联系你,一经查实,本站将立刻删除涉嫌侵权内容。