Why I am here--细谈如何Hadoop重写分块函数，改变分块规则

标签：

• 2015-03-26 11:29

  论文需要用到这一部分技术，但苦于一直没有深入的学习，所以还请您赐教，谢谢！！！qq:945856510,为了交流方便。

   
• 
  阿笨猫 2015-03-26 11:40

  hadoop 我也只是业余用用,不是很精通.
  关于 你的问题 "细谈如何Hadoop重写分块函数，改变分块规则"
  是否而已详细说一下你的需求,看看我是否能帮上忙吧

   
• 
  2015-03-26 11:42

  恩恩，

   
• 
  2015-03-26 11:45

  我的文件是一个按一行一行存储的文件，而且只有完整的一行才是有效数据，而实际上当我把文件上传到HDFS上的时候，它只是非常机械的按64M（默认的大小）一块存储的，这就破坏了我数据的有效性，所以我想是否能重写这个分块函数，在里面加一些规则。。。这是我目前能想到的方法

   
• 
  2015-03-26 11:47

  数据一般格式：keyword deweys
  book 0.1.1.2_0.2.2.1_0.2.3.1
  title 1.2.1.3_7.5.3
  ...
  这样的形式

   
• 
  阿笨猫 2015-03-26 11:51

  我明白你的问题了,这个问题的解决,从来都不是修改block的划分哦.而是修改你的reader的代码哦.
  虽然数据是按照block划分,但是实际你在tt上读取的数据,可不是只能读取到一个block哦,数据读取是可以跨越block界限的.block只是hadoop自己存储的数据分块组织,跟你的使用没关系
  这个hadoop的自带的reader里面就有这样的代码,我待会给你找一下是那个代码,待会去吃饭,你可能得多等一会

   
• 
  2015-03-26 11:52

  好的，我也去吃饭哈…40后能会实验室，谢谢你呀

   
• 
  2015-03-26 11:53

  40分钟后，。。。有点小激动啦

   
• 
  阿笨猫 2015-03-26 12:04

  http://blog.csdn.net/skywalker_only/article/details/42678659
  你阅读下这个博文,
  重点阅读下"FileInputFormat的主要子类有..."后面这段
  对于你的需求,TextInputFormat是可以满足的.
  对于你的疑问,阅读"上述分割InputSplit的逻辑完全是针对大小进行的，那么是否存在将一行记录划分到两个InputSplit中的可能性？" 后面这一段.

  如果还不明白的话,请阅读TextInputFormat,LineRecordReader代码.
  在不明白,就Google,看别人怎么解释的.

  我只能帮你到这里了,剩下的靠你啦

   
• 
  阿笨猫 2015-03-26 13:58

  是否解决了你的疑惑? 你看过那个 hadoop权威指南 没有,可以网上下载一个看看

   
• 
  2015-03-26 14:40

  恩恩，疑惑是解开了，但是就是不知道怎么用KeyValueTextInputFormat或者TextInputFormat（刚刚一直在调试）。里面对这两个类的描述是这样的：
  1、KeyValueTextInputFormat：用于纯文本文件的InputFormat，每行使用分隔字节划分为键和值，该分隔符由参数mapreduce.input.keyvaluelinerecordreader.key.value.separator指定，默认使用\t。如果该分隔符不存在则整行将做为键，值为空。RecordReader为KeyValueLineRecordReader。
  2、TextInputFormat：用于纯文本文件的InputFormat，也是默认的InputFormat。输入文件被分解为行，回车或者换行做为行结束的标记，键为行在文件中的位置，值为行内容。该InputFormat使用LineRecordReader读取InputSplit的内容。
  现在问题就是MapReducer本身是内部调用TextInputFormat的，换句话说就是比如：我现在想用KeyValueTextInputFormat，但是该如何设置（配置）参数呢，让MapReducer知道去调用KeyValueTextInputFormat呢？那本书我没有看过，但是有。我现在就是想知道怎么调用KeyValueTextInputFormat。。。

   
• 
  2015-03-26 14:45

  你忙吧，我要是先弄弄，出来了就告诉你哈，谢谢你啊！！

   
• 
  2015-03-26 18:03

  那个问题已经解决了，主要是在引用包的时候要注意hadoop版本的区别，谢谢你啊

   
• 
  2015-04-13 17:18

  您好！这两天我一直纠结于一个非常奇怪的问题，憋了两天了，实在没有办法，只好来求助于你了.....希望能从你这里获得解决问题的思路。I am really looking forward to your reply,thank you!

   
• 
  阿笨猫 2015-04-13 17:20

  有问题就直接留言啊,你这样多浪费时间?

   
• 
  2015-04-13 17:37

  输入数据格式（给map的数据）：
  John-0.0.0_0.1.0.0.0_0.1.1.1.0_0.1.2.0.0_0.2.0.0.0
  Projects-0.2
  我想得到的输出数据格式：
  Projects-0.2
  John-0.0.0_0.1.0.0.0_0.1.1.1.0_0.1.2.0.0_0.2.0.0.0
  换句话说就是按照单词后面的dewey码的个数从小到大排序，而不是单词的顺序，我的排序程序核心程序如下：
  public static class Map extends Mapper{
  public void map(Text key,Text value,Context context)throws IOException,InterruptedException{
  String[] s=value.toString().trim().split("_");
  context.write(new Text(s.length+""), new Text(key.toString()+"-"+value)); //key与values之间是用"="符号隔开，这在main函数中进行了设置
  }
  }

  public static class Reduce extends Reducer<Text,Text,Text,Text>{	
  	public void reduce(Text key,Iterable<Text> values,Context context)throws IOException,InterruptedException{
  		Iterator<Text> itr = values.iterator();
  		Text t=itr.next();
  		StringTokenizer st = new StringTokenizer(t.toString(),"-");
  		System.out.println("st的大小为："+st.countTokens());
  	}
  }

  控制台输出的结果为：
  15/04/13 17:34:16 INFO mapred.MapTask: data buffer = 79691776/99614720
  15/04/13 17:34:16 INFO mapred.MapTask: record buffer = 262144/327680
  15/04/13 17:34:16 INFO mapred.MapTask: Starting flush of map output
  st的大小为：2
  st的大小为：2
  15/04/13 17:34:17 INFO mapred.MapTask: Finished spill 0
  15/04/13 17:34:17 INFO mapred.TaskRunner: Task:attempt_local_0001_m_000000_0 is done. And is in the process of commiting
  15/04/13 17:34:17 INFO mapred.LocalJobRunner:
  15/04/13 17:34:17 INFO mapred.TaskRunner: Task ‘attempt_local_0001_m_000000_0‘ done.
  15/04/13 17:34:17 INFO mapred.LocalJobRunner:
  15/04/13 17:34:17 INFO mapred.Merger: Merging 1 sorted segments
  15/04/13 17:34:17 INFO mapred.Merger: Down to the last merge-pass, with 1 segments left of total size: 10 bytes
  15/04/13 17:34:17 INFO mapred.LocalJobRunner:
  st的大小为：0

  这让我非常苦恼，真实搞不明白，为什么会出现-----st的大小为：0----这样一个输出，我也不知道到底是那个环节出了问题，还望您指点

   
• 
  阿笨猫 2015-04-13 20:36

  我建议你把map里面的key,value都System.out输出下看看,是不是和你想象的一样?

   
• 
  阿笨猫 2015-04-13 20:37

  再给你一个建议,把上次推荐的 hadoop权威指南 看完,下次书上可以找到答案的,就不回复你了

   
• 
  阿笨猫 2015-04-13 20:39

  最后一个建议,建议你尝试下streaming,针对你之前提的问题,我建议试试streaming.
  http://www.cnblogs.com/luchen927/archive/2012/01/16/2323448.html

   
• 
  2015-04-13 20:45

  恩，好的。我在map里面都输出看了，是有数据的，但是就是不知道reduce为什么最后还蹦出一条数据，那条数据在reduce里面，用System.out怎么都显示不出来，而且更奇怪的是单只是在reduce里面System.out能看到和map里面一样的数据，一加上context.write写的时候，就报越界，而如果我进行if控制之后，context.write正常归约输出的数据也不能写入HDFS中了，。。。。。非常奇怪

   
• 
  阿笨猫 2015-04-13 20:47

  把完整代码贴出来吧,这里放不下,就发你博客里吧.数据文件cat几行出来,全部要正式的情况,我运行下试试

   
• 
  2015-04-13 20:57

  实在不好意思，还没有开通博客，代码只能现将就一下了，

   
• 
  2015-04-13 20:57

  package ile.mapreduce.hqs.www.school;

  import java.io.IOException;
  import java.util.Iterator;
  import java.util.StringTokenizer;

  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.Mapper;
  import org.apache.hadoop.mapreduce.Reducer;
  import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  import org.apache.hadoop.util.Tool;
  import org.apache.hadoop.util.ToolRunner;

  /**

  • @author hzd
  • @version 2015-4-12 14:34
  • 1.功能：对XML的关键字，根据dewey码进行排序
  • 2.现在可以确定的是，‘-‘号作为分割符（其实在解析大文件出现的各种问题，实在是有点让人难于理解）
  • 3.使用了自定义的AW_KeyValueTextInputFormat类进行读取Hadoop上的文件，其实hadoop

  • 本身就有KeyValueTextInputFormat，但是由于版本更新的问题使得接口有所变更，导致函数参数无法正确的对上
    */
    public class AW3_Sort_Deweys implements Tool{

    public static class Map extends Mapper{
    public void map(Text key,Text value,Context context)throws IOException,InterruptedException{
    StringTokenizer st = new StringTokenizer(value.toString(),"\n");
    while(st.hasMoreTokens()){
    String[] s=st.nextToken().split("_");
    context.write(new Text(s.length+""), new Text(key.toString()+"-"+value));
    }
    }
    }

    public static class Reduce extends Reducer{
    public void reduce(Text key,Iterable values,Context context)throws IOException,InterruptedException{

    	Iterator<Text> itr = values.iterator();
    	Text t=itr.next();
    	StringTokenizer st = new StringTokenizer(t.toString(),"-");
    	int i=0;
    	Text k=null,v=null;
    	while(st.hasMoreTokens()){
    		System.out.println(st.nextToken());
    		if(0==i){
    			k = new Text(st.nextToken());
    		}else{
    			v = new Text(st.nextToken());
    		}
    	}
    	context.write(k,v);
    }

    }

    @Override
    public int run(String[] arg0) throws Exception {
    // TODO Auto-generated method stub
    String input = "hdfs://localhost:9000/user/hzd/output-selected/";
    String output = "hdfs://localhost:9000/user/hzd/output-sorted/";

    Path inputDir = new Path(input); 
    Path outputDir = new Path(output); 
    
    Configuration conf=new Configuration();
    
    conf.set("mapred.textoutputformat.separator","-");	//输出的时候也按"-"符号作为分隔符
    
    Job job=new Job(conf,"Sort_Deweys");
    
    job.setInputFormatClass(AW_KeyValueTextInputFormat.class);	//我也是人才啊，这个Hadoop干了这种调包的事情都不说一声
    
    
    job.setJarByClass(AW3_Sort_Deweys.class);
    
    job.setMapperClass(Map.class);
    job.setCombinerClass(Reduce.class);
    job.setReducerClass(Reduce.class);
    
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(Text.class);
    
    FileInputFormat.addInputPath(job, inputDir);
    FileOutputFormat.setOutputPath(job, outputDir);
    
    job.waitForCompletion(true);
    
    return job.isSuccessful()?0:1;

    }

    public static void main(String[] args) throws Exception {
    AW3_Sort_Deweys sort_dewey = new AW3_Sort_Deweys();
    int exit = ToolRunner.run(sort_dewey, args);
    System.exit(exit);
    }

    @Override
    public Configuration getConf() {
    // TODO Auto-generated method stub
    return null;
    }

    @Override
    public void setConf(Configuration arg0) {
    // TODO Auto-generated method stub

    }
    }

   
• 
  2015-04-13 21:04

  http://pan.baidu.com/s/1c0B6efY

   
• 
  2015-04-13 21:04

  用到的两个Java类我放到云盘上了，你看看

   
• 
  阿笨猫 2015-04-13 21:06

  part-r-00000 是输入是吗?

   
• 
  2015-04-13 21:05

  对的

   
• 
  阿笨猫 2015-04-13 21:06

  弱弱的问一句,你注册了博客园,难道没有给你开通博客?

   
• 
  2015-04-13 21:09

  其实我这样就是为了达到下面这种效果
  输入数据格式（给map的数据）：
  John-0.0.0_0.1.0.0.0_0.1.1.1.0_0.1.2.0.0_0.2.0.0.0
  Projects-0.2
  我想得到的输出数据格式：
  Projects-0.2
  John-0.0.0_0.1.0.0.0_0.1.1.1.0_0.1.2.0.0_0.2.0.0.0
  换句话说就是按照单词后面的dewey码的个数从小到大排序，而不是单词的顺序

   
• 
  2015-04-13 21:11

  纯属失误，给你带来不便，还请谅解，我今晚就申请开通

   
• 
  阿笨猫 2015-04-13 21:31

  AW_KeyValueTextInputFormat 你这个类做了啥改动?为啥要做这样的改动?默认的textinputformat为什么不符合你的功能?

   
• 
  2015-04-13 21:36

  是的，因为我要，按"—"符号进行分割

   
• 
  阿笨猫 2015-04-13 21:37

  我囧,你使用默认的textinputformat,然后在map里面你得到的是key(offset),和value(Projects-0.2),你在map函数里自己split好了,这样不行吗?

   
• 
  2015-04-13 21:43

  主要是我要借助rreduce对Dewey码进行排序

   
• 
  阿笨猫 2015-04-13 21:46

  John-0.0.0_0.1.0.0.0_0.1.1.1.0_0.1.2.0.0_0.2.0.0.0
  这个数据,Dewey码是指0.0.0_0.1.0.0.0_0.1.1.1.0_0.1.2.0.0_0.2.0.0.0 吗?
  按照这个进行字典序排序?

   
• 
  2015-04-13 21:43

  如果数据量很大的话，自己写程序进行排序，后果很严重……而如果能用其他方法实现我说的那种顺序，当然是也可以，但是，我就一直苦于为什么在reduce里面会报那种错误

   
• 
  2015-04-13 21:43

  如果数据量很大的话，自己写程序进行排序，后果很严重……而如果能用其他方法实现我说的那种顺序，当然是也可以，但是，我就一直苦于为什么在reduce里面会报那种错误

   
• 
  2015-04-13 21:46

  如果数据量很大的话，自己写程序进行排序，后果很严重……而如果能用其他方法实现我说的那种顺序，当然是也可以，但是，我就一直苦于为什么在reduce里面会报那种错误

   
• 
  2015-04-13 21:45

  后面的是dewey码，但是是按Dewey码的个数的多少进行排序，而不是字典序

   
• 
  2015-04-13 21:49

  忘了说明了，一个关键字，如果有多个Dewey码，那么Dewey之间用"_"隔开

   
• 
  阿笨猫 2015-04-13 21:51

  "按Dewey码的个数的多少进行排序" 这个逻辑你在代码里面那部分做的?

   
• 
  2015-04-13 21:54

  借助map里面算出来的Dewey码的个数，进行排序

   
• 
  阿笨猫 2015-04-13 22:00

  先抛开你说的那个reduce的输出问题.
  你想排序,你现在使用的是hadoop默认的reduce会对key进行排序,不过你知道reduce端对key进行排序,是字典序的,也就意味着顺序可能是,1,11,2,3... ,11个Dewey码的要排在2个Dewey码的前面

   
• 
  2015-04-13 22:03

  恩，这个我知道，我会控制在没9个一组的基础上进行处理

   
• 
  2015-04-13 22:03

  这是我目前的思路

   
• 
  阿笨猫 2015-04-13 22:03

  原来如此,那就好
  今天时间有点晚了,我先下了,建议你先不纠结为啥有哪个输出,先搞定你要的功能吧.我明天给你一个可运行的版本吧

   
• 
  2015-04-13 22:04

  好的，非常感谢，晚安！

   
• 
  阿笨猫 2015-04-13 22:44

  
package test;

  import java.io.IOException;
  import java.util.Iterator;
  import java.util.StringTokenizer;
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.fs.Path;
  import org.apache.hadoop.io.IntWritable;
  import org.apache.hadoop.io.LongWritable;
  import org.apache.hadoop.io.Text;
  import org.apache.hadoop.mapreduce.Job;
  import org.apache.hadoop.mapreduce.Mapper;
  import org.apache.hadoop.mapreduce.Reducer;
  import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  import org.apache.hadoop.util.Tool;
  import org.apache.hadoop.util.ToolRunner;

  public class AW3_Sort_Deweys implements Tool{

  public static class Map extends Mapper<LongWritable,Text,LongWritable,Text>{
  	public void map(LongWritable key,Text value,Context context)throws IOException,InterruptedException{
          String[] kv = value.toString().split("-");
          if (kv.length<2) return;
          String[] ss = kv[1].split("_"); 
  		context.write(new LongWritable(ss.length), value);
  	}
  }
  
  public static class Reduce extends Reducer<LongWritable,Text,Text,Text>{
  	public void reduce(LongWritable key,Iterable<Text> values,Context context)throws IOException,InterruptedException{
  		for(Text v :values)
              context.write(new Text(),v);
  	}
  }
  
  @Override
  public int run(String[] arg0) throws Exception {
  	String input = "/wireless_ha3_xuxm/in1/";  
      String output = "/wireless_ha3_xuxm/out1/";
      Path inputDir = new Path(input); 
      Path outputDir = new Path(output); 
      
  	Configuration conf=new Configuration();
  	Job job=new Job(conf,"Sort_Deweys");
  	job.setJarByClass(AW3_Sort_Deweys.class);
  	job.setMapperClass(Map.class);
  	job.setReducerClass(Reduce.class);
  	
  	FileInputFormat.addInputPath(job, inputDir);
  	FileOutputFormat.setOutputPath(job, outputDir);
  	job.waitForCompletion(true);
  	return job.isSuccessful()?0:1;
  }
  
  public static void main(String[] args) throws Exception {
  	AW3_Sort_Deweys sort_dewey = new AW3_Sort_Deweys();
  	int exit = ToolRunner.run(sort_dewey, args);
  	System.exit(exit);
  }
  
  @Override
  public Configuration getConf() {return null;}
  @Override
  public void setConf(Configuration arg0) {}

  }


   
• 
  阿笨猫 2015-04-13 22:42

  [admin at inc-search-dev-offer3 doc-count]$ hadoop fs -cat /wireless_ha3_xuxm/in1/*
  Ben-0.1.1.2.0_0.1.2.1.0_0.2.0.0.1_0.3.0.0.0_0.3.1.0.0
  John-0.0.0_0.1.0.0.0_0.1.1.1.0_0.1.2.0.0
  Projects-0.2
  [admin at inc-search-dev-offer3 doc-count]$ hadoop fs -cat /wireless_ha3_xuxm/out1/part-r-00000
  Projects-0.2
  John-0.0.0_0.1.0.0.0_0.1.1.1.0_0.1.2.0.0
  Ben-0.1.1.2.0_0.1.2.1.0_0.2.0.0.1_0.3.0.0.0_0.3.1.0.0

   
• 
  阿笨猫 2015-04-13 22:43

  运行的效果是这样的,我也只能帮到你这里了,后面靠你自己了

Why I am here--细谈如何Hadoop重写分块函数，改变分块规则

标签：

原文地址：http://www.cnblogs.com/hzd-zju/p/4423940.html