mapreduce之partition分区

短命女 2022-08-25 12:51 308阅读 0赞

听了超哥的一席课后逐渐明白了partition，记录一下自己的理解！（thanks 超哥）

  1. package partition;
  3. import java.io.IOException;
  4. import java.net.URI;
  6. import org.apache.hadoop.conf.Configuration;
  7. import org.apache.hadoop.fs.FileSystem;
  8. import org.apache.hadoop.fs.Path;
  9. import org.apache.hadoop.io.LongWritable;
  10. import org.apache.hadoop.io.Text;
  11. import org.apache.hadoop.mapreduce.Job;
  12. import org.apache.hadoop.mapreduce.Mapper;
  13. import org.apache.hadoop.mapreduce.Reducer;
  14. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  15. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
  16. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  17. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
  19. /**
  20.  * @ClassName: FlowCount2 
  21.  * @Description: TODO(这里用一句话描述这个类的作用) 
  22.  * @author zhangweixiang
  23.  * @date 2014年3月6日 下午3:27:56
  24.  */
  25. /**
  26.  * 分区的例子必须打成jar运行
  27.  * 用处：    1.根据业务需要，产生多个输出文件
  28.  *         2.多个reduce任务在运行，提高整体job的运行效率
  29.  */
  30. public class FlowCount2 {
  32.     public static final String INPUT_PATH = "hdfs://192.168.0.9:9000/wlan2";
  33.     public static final String OUT_PATH = "hdfs://192.168.0.9:9000/myout";
  35.     public static void main(String[] args) throws Exception {
  36.         Configuration conf = new Configuration();
  37.         Job job = new Job(conf, FlowCount2.class.getSimpleName());
  39.         //指定打包的jar
  40.         job.setJarByClass(FlowCount2.class);
  42.         // 1.1指定输入文件的路径
  43.         FileInputFormat.addInputPath(job, new Path(INPUT_PATH));
  44.         // 指定输入信息的格式化类
  45.         job.setInputFormatClass(TextInputFormat.class);
  47.         // 1.2指定自定义map类
  48.         job.setMapperClass(MyMapper.class);
  49.         // 设置map输出类型
  50.         job.setMapOutputKeyClass(Text.class);
  51.         job.setMapOutputValueClass(FlowWritable.class);
  53.         // 1.3指定分区
  54.         job.setPartitionerClass(MyPartition.class);
  55.         // 设置reduce的任务个数,由于map输出后建立了两个分区，所以应该设置两个reduce任务输出到不同的文件（一个分区对应一个reduce任务）
  56.         job.setNumReduceTasks(2);
  58.         // 1.4排序，分组
  60.         // 1.5规约
  62.         // 2.2指定自定义的reduce类
  63.         job.setReducerClass(MyReduce.class);
  64.         // 设置输出类型
  65.         job.setOutputKeyClass(Text.class);
  66.         job.setOutputValueClass(FlowWritable.class);
  67.         // 设置输出格式化类
  68.         job.setOutputFormatClass(TextOutputFormat.class);
  70.         // 如果输出文件路径存在则删除
  71.         FileSystem fileSystem = FileSystem.get(new URI(OUT_PATH),
  72.                 new Configuration());
  73.         Path path = new Path(OUT_PATH);
  74.         if (fileSystem.exists(path)) {
  75.             fileSystem.delete(path, true);
  76.         }
  78.         // 2.3指定输出路径
  79.         FileOutputFormat.setOutputPath(job, new Path(OUT_PATH));
  81.         // 提交任务
  82.         job.waitForCompletion(true);
  84.     }
  86.     static class MyMapper extends
  87.             Mapper<LongWritable, Text, Text, FlowWritable> {
  88.         @Override
  89.         protected void map(LongWritable key, Text value, Context context)
  90.                 throws IOException, InterruptedException {
  91.             // 分割行
  92.             String[] split = value.toString().split("\t");
  93.             // 获取用户电话号码
  94.             String mobile = "";
  95.             long upPackNum = 0l;
  96.             long downPackNum = 0l;
  97.             long upPayLoad = 0l;
  98.             long downPayLoad = 0l;
  100.             // 符合规范的电话号码
  101.             if (!("".equals(split[2]))) {
  102.                 mobile = split[2];
  103.                 // 获取流量信息
  104.                 if (!("".equals(split[21]))) {
  105.                     upPackNum = Long.parseLong(split[21]);
  106.                 }
  107.                 if (!("".equals(split[22]))) {
  108.                     downPackNum = Long.parseLong(split[22]);
  109.                 }
  110.                 if (!("".equals(split[23]))) {
  111.                     upPayLoad = Long.parseLong(split[23]);
  112.                 }
  113.                 if (!("".equals(split[24]))) {
  114.                     downPayLoad = Long.parseLong(split[24]);
  115.                 }
  117.                 FlowWritable flowWritable = new FlowWritable(upPackNum,
  118.                         downPackNum, upPayLoad, downPayLoad);
  120.                 context.write(new Text(mobile), flowWritable);
  121.             }
  123.         }
  124.     }
  126.     static class MyReduce extends
  127.             Reducer<Text, FlowWritable, Text, FlowWritable> {
  128.         @Override
  129.         protected void reduce(Text k2, Iterable<FlowWritable> v2s,
  130.                 Context context) throws IOException, InterruptedException {
  132.             long upPackNum = 0l;
  133.             long downPackNum = 0l;
  134.             long upPayLoad = 0l;
  135.             long downPayLoad = 0l;
  137.             for (FlowWritable flowWritable : v2s) {
  138.                 upPackNum += flowWritable.upPackNum;
  139.                 downPackNum += flowWritable.downPackNum;
  140.                 upPayLoad += flowWritable.upPayLoad;
  141.                 downPayLoad += flowWritable.downPayLoad;
  142.             }
  144.             FlowWritable flowWritable = new FlowWritable(upPackNum,
  145.                     downPackNum, upPayLoad, downPayLoad);
  147.             context.write(k2, flowWritable);
  148.         }
  149.     }
  152. }
  157. package partition;
  159. import java.io.DataInput;
  160. import java.io.DataOutput;
  161. import java.io.IOException;
  163. import org.apache.hadoop.io.Writable;
  165. /**
  166.  * @ClassName: flowWritable
  167.  * @Description: 自定义类型实现Writable接口，包含四个参数（upPackNum 上行包， downPackNum 下行包，
  168.  *               upPayLoad 发送流量，downPayLoad 下载流量）
  169.  * @author zhangweixiang
  170.  * @date 2014年3月5日 上午11:37:10
  171.  */
  172. public class FlowWritable implements Writable {
  174.     public long upPackNum;
  175.     public long downPackNum;
  176.     public long upPayLoad;
  177.     public long downPayLoad;
  179.     public FlowWritable() {
  180.         // TODO Auto-generated constructor stub
  181.     }
  183.     public FlowWritable(long upPackNum, long downPackNum, long upPayLoad,
  184.             long downPayLoad) {
  185.         this.upPackNum = upPackNum;
  186.         this.downPackNum = downPackNum;
  187.         this.upPayLoad = upPayLoad;
  188.         this.downPayLoad = downPayLoad;
  189.     }
  191.     @Override
  192.     public void write(DataOutput out) throws IOException {
  193.         out.writeLong(upPackNum);
  194.         out.writeLong(downPackNum);
  195.         out.writeLong(upPackNum);
  196.         out.writeLong(downPayLoad);
  197.     }
  199.     @Override
  200.     public void readFields(DataInput in) throws IOException {
  201.         this.upPackNum = in.readLong();
  202.         this.downPackNum = in.readLong();
  203.         this.upPayLoad = in.readLong();
  204.         this.downPayLoad = in.readLong();
  205.     }
  207.     /*
  208.      * (非 Javadoc)
  209.      * 
  210.      * 
  211.      * @return
  212.      * 
  213.      * @see java.lang.Object#toString()
  214.      */
  215.     @Override
  216.     public String toString() {
  217.         return upPackNum + "\t" + downPackNum + "\t" + upPayLoad + "\t"
  218.                 + downPayLoad;
  219.     }
  221. }
  226. package partition;
  228. import org.apache.hadoop.io.Text;
  229. import org.apache.hadoop.mapreduce.lib.partition.HashPartitioner;
  230. /**
  231.  * @ClassName: MyPartition 
  232.  * @Description: 根据电话号码分区，正规号码分区代号为0，非正规号码分区为1（在此建立了两个分区，即会产生两个reduce任务输出到不同的文件0和1）
  233.  * @param K k2(map输出的键), V v2(map输出的值)
  234.  * @author zhangweixiang
  235.  * @date 2014年3月6日 下午3:02:29
  236.  */
  237. public class MyPartition extends HashPartitioner<Text,FlowWritable>{
  238.     @Override
  239.     public int getPartition(Text key, FlowWritable value, int numReduceTasks) {
  240.         int p=0;
  241.         if(key.toString().length()!=11){
  242.             p=1;
  243.         }
  244.         return p;
  245.     }
  246. }

注：必须要达成jar包上传到linux下执行（我开始没有打成jar包直接在eclipse下执行抛了异常）

执行完成后会产生两个文件（part-r-00000和part-r-00001）分别记录不同条件的信息。

eclipse直接运行抛的异常：

（

14/03/06 15:41:13 WARN mapred.LocalJobRunner: job_local_0001
java.io.IOException: Illegal partition for 10.80.203.79 (1)
at org.apache.hadoop.mapred.MapTask$MapOutputBuffer.collect(MapTask.java:1073)
at org.apache.hadoop.mapred.MapTask$NewOutputCollector.write(MapTask.java:691)
at org.apache.hadoop.mapreduce.TaskInputOutputContext.write(TaskInputOutputContext.java:80)
at partition.FlowCount2$MyMapper.map(FlowCount2.java:120)
at partition.FlowCount2$MyMapper.map(FlowCount2.java:1)
at org.apache.hadoop.mapreduce.Mapper.run(Mapper.java:144)
at org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:764)
at org.apache.hadoop.mapred.MapTask.run(MapTask.java:370)
at org.apache.hadoop.mapred.LocalJobRunner$Job.run(LocalJobRunner.java:214)
14/03/06 15:41:14 INFO mapred.JobClient: map 0% reduce 0%
14/03/06 15:41:14 INFO mapred.JobClient: Job complete: job_local_0001
14/03/06 15:41:14 INFO mapred.JobClient: Counters: 0

）

记录超哥的总结：

  1. 分区的例子必须打成jar运行
  2.  * 用处：
  4. *1.根据业务需要，产生多个输出文件
  5.  * 2.多个reduce任务在运行，提高整体job的运行效率

发表评论

表情：
评论列表 （有 0 条评论，308人围观）

还没有评论，来说两句吧...

相关阅读

    相关 hive分区partition介绍

    一、简要概述 1. Hive分区更方便于数据管理，常见的有时间分区和业务分区。 二、hive分区原理 通过实例来理解Hive分区的原理： （一）多分区操作： 1.

    快来打我*快来打我*/ 0 赞/ 323 阅读

    相关 Hive partition 分区表

    分区表实际上就是对应一个HDFS文件系统上的独立的文件夹，该文件夹下是该分区所有的数据文件。Hive中的分区就是分目录，把一个大的数据集根据业务需要分割成小的数据集。在查询时通

    迈不过友情╰迈不过友情╰/ 0 赞/ 432 阅读

    相关 MySQL数据库：分区Partition

    一、分区： 分区就是将表的数据按照特定规则存放在不同的区域，也就是将表的数据文件分割成多个小块，在查询数据的时候，只要知道数据数据存储在哪些区域，然后直接在对应的区域进行

    ╰半夏微凉°╰半夏微凉°/ 0 赞/ 408 阅读