Hadoop学习(4)-mapreduce的一些注意事项
Hadoop学习(4)-mapreduce的一些注意事项
关于mapreduce的一些注意细节
如果把mapreduce程序打包放到了liux下去运行,
命令java –cp xxx.jar 主类名
如果报错了,说明是缺少相关的依赖jar包
用命令hadoop jar xxx.jar 类名因为在集群机器上用 hadoop jar xx.jar mr.wc.JobSubmitter 命令来启动客户端main方法时,hadoop jar这个命令会将所在机器上的hadoop安装目录中的jar包和配置文件加入到运行时的classpath中
那么,我们的客户端main方法中的new Configuration()语句就会加载classpath中的配置文件,自然就有了
fs.defaultFS 和 mapreduce.framework.name 和 yarn.resourcemanager.hostname 这些参数配置
会把本地hadoop的相关的所有jar包都会引用
Mapreduce也有本地的job运行,就是可以不用提交到yarn上,可以以单机的模式跑一边以多个线程模拟也可以。
就是如果不管在Linux下还是windows下,提交job都会默认的提交到本地去运行,
如果在linux默认提交到yarn上运行,需要写配置文件hadoop/etc/mapred-site.xml文件
mapreduce.framework.name
yarn
Key,value对,如果是自己的类的话,那么这个类要实现Writable,同时要把你想序列化的数据转化成二进制,然后放到重写方法wirte参数的DataOutput里面,另一个readFields重写方法是用来反序列化用的,
注意反序列化的时候,会先拿这个类的无参构造方法构造出一个对象出来,然后再通过readFields方法来复原这个对象。
DataOutput也是一种流,只不过是hadoop的在封装,自己用的时候,里面需要加个FileOutputStream对象
DataOutput写字符串的时候要用writeUTF(“字符串”),他这样编码的时候,会在字符串的前面先加上字符串的长度,这是考虑到字符编码对其的问题,hadoop解析的时候就会先读前面两个字节,看一看这个字符串有多长,不然如果用write(字符串.getBytes())这样他不知道这个字符串到底有多少个字节。
在reduce阶段,如果把一个对象写到hdfs里面,那么会调用字符串的toString方法,你可以重写这个类的toString方法
举例,下面这个类就可以在hadoop里序列化
package mapreduce2;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import org.apache.hadoop.hdfs.client.HdfsClientConfigKeys.Write;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.util.Waitable;
public class FlowBean implements Writable {
private int up;//上行流量 private int down;//下行流量 private int sum;//总流量 private String phone;//电话号 public FlowBean(int up, int down, String phone) { this.up = up; this.down = down; this.sum = up + down; this.phone = phone; } public int getUp() { return up; } public void setUp(int up) { this.up = up; } public int getDown() { return down; } public void setDown(int down) { this.down = down; } public int getSum() { return sum; } public void setSum(int sum) { this.sum = sum; } public String getPhone() { return phone; } public void setPhone(String phone) { this.phone = phone; } @Override public void readFields(DataInput di) throws IOException { //注意这里读的顺序要和写的顺序是一样的 this.up = di.readInt(); this.down = di.readInt(); this.sum = this.up + this.down; this.phone = di.readUTF(); } @Override public void write(DataOutput Do) throws IOException { Do.writeInt(this.up); Do.writeInt(this.down); Do.writeInt(this.sum); Do.writeUTF(this.phone); } @Override public String toString() { return "电话号"+this.phone+" 总流量"+this.sum; }
}
当所有的reduceTask都运行完之后,还会调用一个cleanup方法
应用练习:统计一个页面访问总量为n条的数据
方案一:只用一个reducetask,利用cleanup方法,在reducetask阶段,先不直接放到hdfs里面,而是存到一个Treemap里面
再在reducetask结束后,在cleanup里面通过把Treemap里面前五输出到HDFS里面;
package cn.edu360.mr.page.topn;
public class PageCount implements Comparable{
private String page; private int count; public void set(String page, int count) { this.page = page; this.count = count; } public String getPage() { return page; } public void setPage(String page) { this.page = page; } public int getCount() { return count; } public void setCount(int count) { this.count = count; } @Override public int compareTo(PageCount o) { return o.getCount()-this.count==0?this.page.compareTo(o.getPage()):o.getCount()-this.count; }
}
map类
import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
public class PageTopnMapper extends Mapper{
@Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line = value.toString(); String[] split = line.split(" "); context.write(new Text(split[1]), new IntWritable(1)); }
}
reduce类
package cn.edu360.mr.page.topn;
import java.io.IOException;
import java.util.Map.Entry;
import java.util.Set;
import java.util.TreeMap;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
public class PageTopnReducer extends Reducer{
TreeMap<PageCount, Object> treeMap = new TreeMap<>(); @Override protected void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException { int count = 0; for (IntWritable value : values) { count += value.get(); } PageCount pageCount = new PageCount(); pageCount.set(key.toString(), count); treeMap.put(pageCount,null); } @Override protected void cleanup(Context context) throws IOException, InterruptedException { Configuration conf = context.getConfiguration();
//可以在cleanup里面拿到configuration,从里面读取要拿前几条数据
int topn = conf.getInt("top.n", 5); Set<Entry<PageCount, Object>> entrySet = treeMap.entrySet(); int i= 0; for (Entry<PageCount, Object> entry : entrySet) { context.write(new Text(entry.getKey().getPage()), new IntWritable(entry.getKey().getCount())); i++; if(i==topn) return; } }
}
然后jobSubmit类,注意这个要设定Configuration,这里面有几种方法
第一种是加载配置文件
Configuration conf = new Configuration(); conf.addResource("xx-oo.xml");
然后再在xx-oo.xml文件里面写
<property> <name>top.n</name> <value>6</value> </property>
第二种方式
//通过直接设定
conf.setInt("top.n", 3); //通过对java主程序 直接传进来的参数 conf.setInt("top.n", Integer.parseInt(args[0]));
第三种方式通过获取配置文件参数
Properties props = new Properties();
props.load(JobSubmitter.class.getClassLoader().getResourceAsStream("topn.properties")); conf.setInt("top.n", Integer.parseInt(props.getProperty("top.n")));
然后再在topn.properties里面配置参数
top.n=5
subsubmit类,默认在本机模拟运行
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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;
public class JobSubmitter {
public static void main(String[] args) throws Exception { /** * 通过加载classpath下的*-site.xml文件解析参数 */ Configuration conf = new Configuration(); conf.addResource("xx-oo.xml"); /** * 通过代码设置参数 */ //conf.setInt("top.n", 3); //conf.setInt("top.n", Integer.parseInt(args[0])); /** * 通过属性配置文件获取参数 */ /*Properties props = new Properties(); props.load(JobSubmitter.class.getClassLoader().getResourceAsStream("topn.properties")); conf.setInt("top.n", Integer.parseInt(props.getProperty("top.n")));*/ Job job = Job.getInstance(conf); job.setJarByClass(JobSubmitter.class); job.setMapperClass(PageTopnMapper.class); job.setReducerClass(PageTopnReducer.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(IntWritable.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); FileInputFormat.setInputPaths(job, new Path("F:\\mrdata\\url\\input")); FileOutputFormat.setOutputPath(job, new Path("F:\\mrdata\\url\\output")); job.waitForCompletion(true); }
}
额外java知识点补充
Treemap,放进去的东西会自动排序
两种Treemap的自定义方法,第一种是传入一个Comparator
public class TreeMapTest {
public static void main(String[] args) { TreeMap<FlowBean, String> tm1 = new TreeMap<>(new Comparator<FlowBean>() { @Override public int compare(FlowBean o1, FlowBean o2) { //如果两个类总流量相同的会比较电话号 if( o2.getAmountFlow()-o1.getAmountFlow()==0){ return o1.getPhone().compareTo(o2.getPhone()); } //如果流量不同,就按从小到大的顺序排序 return o2.getAmountFlow()-o1.getAmountFlow(); } }); FlowBean b1 = new FlowBean("1367788", 500, 300); FlowBean b2 = new FlowBean("1367766", 400, 200); FlowBean b3 = new FlowBean("1367755", 600, 400); FlowBean b4 = new FlowBean("1367744", 300, 500); tm1.put(b1, null); tm1.put(b2, null); tm1.put(b3, null); tm1.put(b4, null); //treeset的遍历 Set<Entry<FlowBean,String>> entrySet = tm1.entrySet(); for (Entry<FlowBean,String> entry : entrySet) { System.out.println(entry.getKey() +"\t"+ entry.getValue()); } }
}
第二种是在这个类中,实现一个Comparable接口
package cn.edu360.mr.page.topn;
public class PageCount implements Comparable{
private String page; private int count; public void set(String page, int count) { this.page = page; this.count = count; } public String getPage() { return page; } public void setPage(String page) { this.page = page; } public int getCount() { return count; } public void setCount(int count) { this.count = count; } @Override public int compareTo(PageCount o) { return o.getCount()-this.count==0?this.page.compareTo(o.getPage()):o.getCount()-this.count; }
}
原文地址https://www.cnblogs.com/wpbing/archive/2019/07/25/11242866.html
低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。
持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。
转载内容版权归作者及来源网站所有,本站原创内容转载请注明来源。
- 上一篇
【降价信息】云HBase X-Pack最高降价31%,再次释放大数据处理红利
一、降价时间:北京时间7月16日降价内容:阿里云中国站云HBase X-Pack中的Spark服务将于2019年7月16日进行价格下调。本次降价策略覆盖国内北京、上海、杭州、深圳、张北。具体优惠详见下表: 规格 降幅 4CPU 8G 31% 8CPU 16G 31% 16CPU 32G 31% 32CPU 64G 31% 4CPU 16G 11% 8CPU 32G 11% 16CPU 64G 11% 32CPU 128G 11% 二、X-Pack Spark服务X-Pack Spark为HBase、RDS、MongoDB、Redis等服务提供复杂分析、流式处理及入库、机器学习等能力,从而更好的解决大数据处理等场景问题。帮助文档、购买入口
- 下一篇
kafka API consumer
kafka API consumer1.kafka consumer流程1.1.在启动时或者协调节点故障转移时,消费者发送ConsumerMetadataRequest给bootstrap brokers列表中的任意一个brokers。在ConsumerMetadataResponse中,它接收消费者对应的消费组所属的协调节点的位置信息。 1.2.消费者连接协调节点,并发送HeartbeatRequest。如果返回的HeartbeatResponse中返回IllegalGeneration错误码,说明协调节点已经在初始化。消费者就会停止抓取数据,提交offsets,发送JoinGroupRequest给协调节点。在JoinGroupResponse,它接收消费者应该拥有的topic-partitions列表以及当前消费组的新的generation编号。这个时候消费组管理已经完成,消费者就可以开始抓取数据,并为它拥有的partitions提交offsets。 1.3.如果HeartbeatResponse没有错误返回,消费者会从它上次拥有的partitions列表继续抓取数据,这个过程是...
相关文章
文章评论
共有0条评论来说两句吧...
文章二维码
点击排行
推荐阅读
最新文章
- CentOS6,CentOS7官方镜像安装Oracle11G
- Red5直播服务器,属于Java语言的直播服务器
- Hadoop3单机部署,实现最简伪集群
- CentOS7,CentOS8安装Elasticsearch6.8.6
- CentOS7安装Docker,走上虚拟化容器引擎之路
- CentOS7编译安装Gcc9.2.0,解决mysql等软件编译问题
- Windows10,CentOS7,CentOS8安装MongoDB4.0.16
- CentOS7编译安装Cmake3.16.3,解决mysql等软件编译问题
- SpringBoot2整合Redis,开启缓存,提高访问速度
- CentOS6,7,8上安装Nginx,支持https2.0的开启