MapReduce代码

WordCount

数据准备：

a.txt

lxy lxy
lxy zhang
wsoossj liagn
guui
liang
liagn

代码（在idea中创建一个Maven工程）：

mapper：

package com.lxy.mr.wordcount.thi;import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;import java.io.IOException;/*** Author: Liang_Xinyu* Date: 24/05/13* Time: 18:53*/
public class WCMapper extends Mapper<LongWritable, Text,Text,LongWritable> {@Overrideprotected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, LongWritable>.Context context) throws IOException, InterruptedException {String line = value.toString();for (String word : line.split(" ")) {Text outKey = new Text();outKey.set(word);LongWritable outValue = new LongWritable();outValue.set(1);context.write(outKey,outValue);}}
}

reducer

package com.lxy.mr.wordcount.thi;import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mapreduce.Reducer;import javax.xml.soap.Text;
import java.io.IOException;/*** Author: Liang_Xinyu* Date: 24/05/13* Time: 18:59*/
public class WCReducer extends Reducer<Text, LongWritable,Text,LongWritable> {@Overrideprotected void reduce(Text key, Iterable<LongWritable> values, Reducer<Text, LongWritable, Text, LongWritable>.Context context) throws IOException, InterruptedException {long sum = 0;for (LongWritable word : values) {long l = word.get();sum += l;}LongWritable outValue = new LongWritable();outValue.set(sum);context.write(key,outValue);}
}

driver

在本地运行的代码

就是右键运行

package com.lxy.mr.wordcount;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;import java.io.IOException;/*** Author: Liang_Xinyu* Date: 24/05/13* Time: 6:25* 在本地运行*/
public class WCDriver {public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {Job job = Job.getInstance(new Configuration());//2.给Job设置参数 -- 包含mapper和reducer类//2.1关联本Driver程序的jar---本地运行的不需要设置 在集群上运行必须设置job.setJarByClass(WCDriver.class);//2.2设置Mapper和Reducer类job.setMapperClass(WCMapper.class);job.setReducerClass(WCReducer.class);//2.3设置Mapper输出的key,value的类型job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(LongWritable.class);//2.4设置最终输出的key,value的类型-在这是Reducer输出的key value的类型job.setOutputKeyClass(Text.class);job.setOutputValueClass(LongWritable.class);//2.5设置输入和输出路径//设置输入路径--数据所在的路径FileInputFormat.setInputPaths(job,new Path("D:\\io\\input"));//设置输出路径-运算的结果所放的路径-该路径一定不能存在否则报错FileOutputFormat.setOutputPath(job,new Path("D:\\io\\output"));//3.提交Job/*boolean waitForCompletion(boolean verbose)verbose : 是否打印执行的进度返回值 : 如果Job执行成功返回true否则返回false*/boolean b = job.waitForCompletion(true);System.out.println("b========" + b);}
}
在集群运行的代码
在集群上运行Job
1.修改代码 - 输入路径和输出路径从main方法读取
2.打jar包
3.将jar包放到一个有权限的目录中（我放到了家目录）
4.在HDFS上准备数据
5.运行jar包 ： hadoop jar xxx.jar 全类名 参数1 参数2
6.查看hadoop103:8088 是否有Job提交
7.查看hadoop102:9870 查看结果

package com.lxy.mr.wordcount;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;import java.io.IOException;/*** Author: Liang_Xinyu* Date: 24/05/14* Time: 7:45* 在HDFS上提交运行*/
public class WCDemo3 {public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {//创建Job实例Job job = Job.getInstance(new Configuration());//2.给Job设置参数 -- 包含mapper和reducer类//2.1关联本Driver程序的jar---本地运行的不需要设置 在集群上运行必须设置job.setJarByClass(WCDemo3.class);//2.2设置Mapper和Reducer类job.setMapperClass(WCMapper.class);job.setReducerClass(WCReducer.class);//2.3设置Mapper输出的key,value的类型job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(LongWritable.class);//2.4设置最终输出的key,value的类型-在这是Reducer输出的key value的类型job.setOutputKeyClass(Text.class);job.setOutputValueClass(LongWritable.class);//2.5设置输入和输出路径//设置输入路径--数据所在的路径FileInputFormat.setInputPaths(job,new Path(args[0]));FileOutputFormat.setOutputPath(job,new Path(args[1]));System.out.println("b = " + job.waitForCompletion(true));}
}

在集群运行要在VMVare WorkStation或XShell敲命令！！！

  • 先打jar包在这里插入图片描述
  • 再把jar包拖集群里
    在这里插入图片描述
  • 然后敲命令

hadoop jar xxx.jar 全类名 参数1 参数2
这是我的：hadoop jar MRDemo-1.0-SNAPSHOT.jar com.lxy.mr.wordcount.WCDemo3 /input /output
全类名这样获取：
在这里插入图片描述
我们在代码里设置的参数1和参数2分别是
集群上的输入路径和输出路径
这是输入路径
在这里插入图片描述
输出路径一定不要存在，否则会报错，空着就好
在这里插入图片描述
我在执行中遇到了ClassNotFoundException

就把代码中的类名换成全类名解决了，一般是不需要的
在这里插入图片描述

用Windows向集群提交

从本地向集群提交 – 知道即可
1.配置参数
//设置在集群运行的相关参数-设置HDFS,NAMENODE的地址
conf.set(“fs.defaultFS”, “hdfs://hadoop102:8020”);
//指定MR运行在Yarn上
conf.set(“mapreduce.framework.name”,“yarn”);
//指定MR可以在远程集群运行
conf.set(“mapreduce.app-submission.cross-platform”,“true”);
//指定yarn resourcemanager的位置
conf.set(“yarn.resourcemanager.hostname”,“hadoop103”);
2.输入路径和输出路径从main方法读取
3.打jar包
4. //注释掉下面代码
//job.setJarByClass(WCDriver3.class);
//添加如下代码 - 设置jar包路径
job.setJar(“jar包路径”);

5.在右上角点击EditConfigurations -> 左边选类名（没有先运行一次该类）右面选modify Options -> add vm options->  在 vm options的输入框中添加: -DHADOOP_USER_NAME=lxy在 program Arguments添加 ：hdfs://hadoop102:8020/input hdfs://hadoop102:8020/output33

package com.lxy.mr.wordcount;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;import java.io.IOException;/*** Author: Liang_Xinyu* Date: 24/05/14* Time: 1:01* 从本地向集群提交*/
public class WCDriver2 {public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {//        Job job = Job.getInstance(new Configuration());Configuration conf = new Configuration();//设置在集群运行的相关参数-设置HDFS,NAMENODE的地址conf.set("fs.defaultFS", "hdfs://hadoop102:8020");//指定MR运行在Yarn上conf.set("mapreduce.framework.name","yarn");//指定MR可以在远程集群运行conf.set("mapreduce.app-submission.cross-platform","true");//指定yarn resourcemanager的位置conf.set("yarn.resourcemanager.hostname","hadoop103");Job job = Job.getInstance(conf);//2.给Job设置参数 -- 包含mapper和reducer类//2.1关联本Driver程序的jar---本地运行的不需要设置 在集群上运行必须设置
//        job.setJarByClass(WCDriver2.class);job.setJar("D:\\words\\idea\\MRDemo\\target\\MRDemo-1.0-SNAPSHOT.jar");//2.2设置Mapper和Reducer类job.setMapperClass(WCMapper.class);job.setReducerClass(WCReducer.class);//2.3设置Mapper输出的key,value的类型job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(LongWritable.class);//2.4设置最终输出的key,value的类型-在这是Reducer输出的key value的类型job.setOutputKeyClass(Text.class);job.setOutputValueClass(LongWritable.class);//2.5设置输入和输出路径//设置输入路径--数据所在的路径 yarn:args[0]FileInputFormat.setInputPaths(job,new Path(args[0]));//设置输出路径-运算的结果所放的路径-该路径一定不能存在否则报错FileOutputFormat.setOutputPath(job,new Path(args[1]));//3.提交Job/*boolean waitForCompletion(boolean verbose)verbose : 是否打印执行的进度返回值 : 如果Job执行成功返回true否则返回false*/System.out.println("b ==== " + job.waitForCompletion(true));}
}

向集群提交的配置在这里修改：
在这里插入图片描述
在这里插入图片描述
我们之前的代码都是setJarByClass，现在是setJar，要写绝对路径，绝对路径从这里获取：
在这里插入图片描述

流量统计

数据准备

phone_data.txt

1	13736230513	192.196.100.1	www.atguigu.com	2481	24681	200
2	13846544121	192.196.100.2			264	0	200
3 	13956435636	192.196.100.3			132	1512	200
4 	13966251146	192.168.100.1			240	0	404
5 	18271575951	192.168.100.2	www.atguigu.com	1527	2106	200
6 	84188413	192.168.100.3	www.atguigu.com	4116	1432	200
7 	13590439668	192.168.100.4			1116	954	200
8 	15910133277	192.168.100.5	www.hao123.com	3156	2936	200
9 	13729199489	192.168.100.6			240	0	200
10 	13630577991	192.168.100.7	www.shouhu.com	6960	690	200
11 	15043685818	192.168.100.8	www.baidu.com	3659	3538	200
12 	15959002129	192.168.100.9	www.atguigu.com	1938	180	500
13 	13560439638	192.168.100.10			918	4938	200
14 	13470253144	192.168.100.11			180	180	200
15 	13682846555	192.168.100.12	www.qq.com	1938	2910	200
16 	13992314666	192.168.100.13	www.gaga.com	3008	3720	200
17 	13509468723	192.168.100.14	www.qinghua.com	7335	110349	404
18 	18390173782	192.168.100.15	www.sogou.com	9531	2412	200
19 	13975057813	192.168.100.16	www.baidu.com	11058	48243	200
20 	13768778790	192.168.100.17			120	120	200
21 	13568436656	192.168.100.18	www.alibaba.com	2481	24681	200
22 	13568436656	192.168.100.19			1116	954	200

FlowBean

让当前类的对象可以在Hadoop中序列化和反序列化

FlowBean这个类是一个JavaBean

Writable序列化步骤
1.自定义一个类(作为MR中的key或value)并实现Writable
2.实现write和readFields方法
3.在write方法中写序列化时要做的事情
在readFields方法中写反序列化时要做的事情
4.注意：反序列化时的顺序要和序列化时的顺序保持一致

package com.lxy.mr.exer.writable.mr01;import org.apache.hadoop.io.Writable;import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;/*** Author: Liang_Xinyu* Date: 24/05/14* Time: 7:26*/
public class FLowBean implements Writable {private long upFlow;private long downFlow;private long sumFlow;public long getUpFlow() {return upFlow;}public void setUpFlow(long upFlow) {this.upFlow = upFlow;}public long getDownFlow() {return downFlow;}public void setDownFlow(long downFlow) {this.downFlow = downFlow;}public long getSumFlow() {return sumFlow;}public void setSumFlow(long sumFlow) {this.sumFlow = sumFlow;}public FLowBean() {}public FLowBean(long upFlow, long downFlow, long sumFlow) {this.upFlow = upFlow;this.downFlow = downFlow;this.sumFlow = sumFlow;}public FLowBean(long upFlow, long downFlow) {this.upFlow = upFlow;this.downFlow = downFlow;this.sumFlow = this.getUpFlow() + this.getDownFlow();}@Overridepublic void write(DataOutput dataOutput) throws IOException {dataOutput.writeLong(downFlow);dataOutput.writeLong(upFlow);dataOutput.writeLong(sumFlow);}@Overridepublic void readFields(DataInput dataInput) throws IOException {downFlow = dataInput.readLong();upFlow = dataInput.readLong();sumFlow = dataInput.readLong();}@Overridepublic String toString() {return "FLowBean{" +"upFlow=" + upFlow +", downFlow=" + downFlow +", sumFlow=" + sumFlow +'}';}
}

Mapper

package com.lxy.mr.exer.writable.mr01;import com.lxy.mr.writable.mr01.FlowBean;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;import java.io.IOException;/*** Author: Liang_Xinyu* Date: 24/05/14* Time: 20:50*/public class FlowMapper extends Mapper<LongWritable, Text,Text, FlowBean> {@Overrideprotected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, FlowBean>.Context context) throws IOException, InterruptedException {//1.切割数据String[] info = value.toString().split("\t");//2.封装key和value (  key(13736230513)  value(2481,24681,2481+24681) )Text outKey = new Text();outKey.set(info[1]);FlowBean outValue = new FlowBean();outValue.setUpFlow(Long.parseLong(info[info.length - 3]));outValue.setDownFlow(Long.parseLong(info[info.length - 2]));outValue.setSumFlow(outValue.getUpFlow() + outValue.getDownFlow());//3.将key,value写出去context.write(outKey,outValue);}
}

Reducer

该类是在reduce阶段被ReduceTask调用 用来实现ReduceTask要实现的业务逻辑代码Reducer<KEYIN, VALUEIN, KEYOUT, VALUEOUT>第一组：KEYIN ：读取的key的类型（map写出的key的类型）-在这是手机号的类型VALUEIN ：读取的value的类型（map写出的value的类型）-在这是FlowBean的类型第二组：KEYOUT : 写出的key的类型-在这是手机号的类型VALUEOUT ：写出的value的类型 - 在这是FlowBean的类型

package com.lxy.mr.exer.writable.mr01;import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;import java.io.IOException;/*** Author: Liang_Xinyu* Date: 24/05/14* Time: 20:50*/
public class FlowReducer extends Reducer<Text, FLowBean,Text,FLowBean> {/*** 在reduce方法中用来实现需要在ReduceTask中实现的功能* reduce方法在被循环调用每调用一次传入一组数据* @param key 读取的key - 单词* @param values 放了所有的value - 单词的数量* @param context 上下文在这用来将key,value写出去* @throws IOException* @throws InterruptedException*/@Overrideprotected void reduce(Text key, Iterable<FLowBean> values, Reducer<Text, FLowBean, Text, FLowBean>.Context context) throws IOException, InterruptedException {long sumDownFlow = 0;long sumUpFlow = 0;for (FLowBean value : values) {sumUpFlow += value.getUpFlow();sumDownFlow += value.getDownFlow();}FLowBean outValue = new FLowBean(sumUpFlow, sumDownFlow);context.write(key,outValue);}
}

Driver

右键运行即可

package com.lxy.mr.exer.writable.mr01;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;import java.io.IOException;/*** Author: Liang_Xinyu* Date: 24/05/14* Time: 7:27*/
public class FlowDriver{public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {//1.创建Job实例Job job = Job.getInstance(new Configuration());//2.给Job设置参数 -- 包含mapper和reducer类//2.1关联本Driver程序的jar---本地运行的不需要设置 在集群上运行必须设置job.setJarByClass(FlowDriver.class);//2.2设置Mapper和Reducer类job.setMapperClass(FlowMapper.class);job.setReducerClass(Reducer.class);//2.3设置Mapper输出的key,value的类型job.setMapOutputKeyClass(Text.class);job.setOutputValueClass(FLowBean.class);//2.4设置最终输出的key,value的类型-在这是Reducer输出的key value的类型job.setOutputKeyClass(Text.class);job.setOutputValueClass(FLowBean.class);//2.5设置输入和输出路径//设置输入路径--数据所在的路径FileInputFormat.setInputPaths(job,new Path("D:\\io\\input2"));//设置输出路径-运算的结果所放的路径-该路径一定不能存在否则报错FileOutputFormat.setOutputPath(job,new Path("D:\\io\\output2"));//3.提交Jobjob.waitForCompletion(true);}
}

结果

在这里插入图片描述

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.hqwc.cn/news/701283.html

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

Python NumPy数组的创建方法

Python NumPy数组的创建方法

Numpy是Python中科学计算的基础包&#xff0c;其核心对象就是ndarray&#xff08;n维数组&#xff09;。利用数组可以快速执行逻辑&#xff0c;形状操作&#xff0c;统计和傅里叶变换等运算&#xff0c;其效率比Python原生的数组效率更高。因此使用Numpy的第一件事就是创建Nump…
阅读更多...
四川景源畅信：抖音有哪些可以做的副业？

四川景源畅信：抖音有哪些可以做的副业？

抖音作为当前最受欢迎的短视频平台之一&#xff0c;其巨大的流量和用户基础为许多人提供了副业的机会。那么&#xff0c;在抖音上可以做哪些副业呢? 一、内容创作与推广 利用抖音平台进行内容创作是最直接的副业方式。无论是搞笑短剧、生活分享还是专业知识普及&#xff0c;只…
阅读更多...
具身智能论文（四）

具身智能论文（四）

目录 1. Alexa Arena: A User-Centric Interactive Platform for Embodied AI2. EDGI: Equivariant Diffusion for Planning with Embodied Agents3. Efficient Policy Adaptation with Contrastive Prompt Ensemble for Embodied Agents4. Egocentric Planning for Scalable E…
阅读更多...
【综述】人工智能、机器学习、深度学习

【综述】人工智能、机器学习、深度学习

文章目录 前言 概念 算法 训练 性能 应用 参考资料 前言 见《初试人工智能》 概念 人工智能系统&#xff08;artifieial intelligence system&#xff09;&#xff0c;针对人类定义的给定目标&#xff0c;产生诸如内容、预测、推荐或决策等输出的一类工程系统。该工程系…
阅读更多...
简单的mysql主从复制搭建

简单的mysql主从复制搭建

文章目录 准备工作用Docker安装MySQL主库配置【192.168.13.32】从库配置【192.168.13.108】小结 准备工作 用虚拟机提前准备两台服务器&#xff0c;并且在服务器中分别安装好MySQL&#xff0c;服务器的信息如下&#xff1a; 数据库IP主节点192.168.13.32从节点192.168.13.108…
阅读更多...
【前端】打砖块游戏：实现细节介绍

【前端】打砖块游戏：实现细节介绍

打砖块游戏：实现细节介绍 在本文中，我将详细介绍如何使用HTML、CSS和JavaScript技术构建一个简单的打砖块游戏。我们将重点讨论游戏的三个核心技术方面：碰撞检测、画图和事件监听。 完整代码我放在：github可以直接拉取代码测试。 游戏概览 打砖块游戏中，玩家通过控制底…
阅读更多...
vue-cropper裁剪图片 vue

vue-cropper裁剪图片 vue

效果图 1.配置环境 npm install vue-cropper 2.代码 <template><div class"cropper-content"><div class"cropper-box"><div class"cropper"><vue-cropper ref"cropper" :img"option.img" :…
阅读更多...
C语言中数组与指针的区别

C语言中数组与指针的区别

一. 简介 本文学习了 C语言中数组与指针的区别。这样的话&#xff0c;可以在编写C代码时规避掉出错的问题。 二. C语言中数组与指针的区别 1. 数组 定义字符串数组时&#xff0c;必须让编译器知道需要多少空间。 一种方法是用足够空间的数组存储字符串。例如如下&#xf…
阅读更多...
【Vue】更换vue-element-admin左侧 logo

【Vue】更换vue-element-admin左侧 logo

准备&#xff1a;目标svg格式的 logo&#xff0c;并将目标logo命名为 vuejs-fill.svg替换路径&#xff1a;/icons 文件夹下&#xff0c;覆盖掉原本的 vuejs-fill.svg 原因&#xff1a;配置项的logo设置的是 vuejs-fill
阅读更多...
Linux中如何配置虚拟机网络（NAT方法）

Linux中如何配置虚拟机网络（NAT方法）

首先我们要在Linux中找到配置文件的路径/etc/sysconfig/network-scripts/&#xff0c;然后找到配置文件的名称ifcfg-xxx&#xff08;如&#xff1a;ifcfg-ens33&#xff09;&#xff0c;然后打开这个文件内 容如下&#xff1a; TYPEEthernet # 指定网卡类型是以太网 BOOTPROT…
阅读更多...
【ARM Cortex-M 系列 2.3 -- Cortex-M7 Debug event 详细介绍】

【ARM Cortex-M 系列 2.3 -- Cortex-M7 Debug event 详细介绍】

请阅读【嵌入式开发学习必备专栏】 文章目录 Cortex-M7 Debug eventDebug events Cortex-M7 Debug event 在ARM Cortex-M7架构中&#xff0c;调试事件&#xff08;Debug Event&#xff09;是由于调试原因而触发的事件。一个调试事件会导致以下几种情况之一发生&#xff1a; 进…
阅读更多...
高级DBA手把手教你达梦8国产数据库级联更新语句用MergeInto合并代替方法（达梦官方手册无此内容）

高级DBA手把手教你达梦8国产数据库级联更新语句用MergeInto合并代替方法（达梦官方手册无此内容）

高级DBA手把手教你达梦8国产数据库级联更新语句用MergeInto合并代替方法&#xff08;达梦官方手册无此内容&#xff09; 一、传统级联更新语句例子 举例&#xff1a; 表 1&#xff1a;T1 字段名类型A时间类型B字符类型C字符类型D字符类型E字符类型 表 2&#xff1a;T2 字…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023