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文件

<name>mapreduce.framework.name</name>

</property>

</configuration>

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<PageCount>{

    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<LongWritable, Text, Text, IntWritable>{

    @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<Text, IntWritable, Text, IntWritable>{

    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文件里面写

<configuration>

    <property>

        <name>top.n</name>

        <value>6</value>

    </property>

</configuration>

第二种方式

　　　　//通过直接设定

        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);

    }

}

有时一个任务一个mapreduce是完成不了的，有可能会拆分成两个或多个mapreduce

map阶段会有自己的排序机制，比如一组数据(a,1),(b,1),(a,1),(c,1),他会先处理key为1的一组数据，

这个排序机制我们也可以自己去实现，要对这个类实现Comparable接口，然后重写compareTo方法。

但要注意这个排序机制只是对于一个reducetask来说的，如果有多个的话，只会得到局部排序。

如果要多个reducetask的话，我们就需要控制数据的分发规则，这样虽然是会生成多个排序后的文件，但这些文件整体上依然是有序的。因为我们控制了每一个reducetask处理数据的范围。

额外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<PageCount>{

    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;

    }

}

巴特西

Hadoop学习(4)-mapreduce的一些注意事项

最新文章

热门文章