Hadoop添加节点的方法

ny3259

　　自己实际添加节点过程：
1. 先在slave上配置好环境，包括ssh，jdk，相关config，lib，bin等的拷贝；
2. 将新的datanode的host加到集群namenode及其他datanode中去；
3. 将新的datanode的ip加到master的conf/slaves中；
4. 重启cluster,在cluster中看到新的datanode节点；
5. 运行bin/start-balancer.sh，这个会很耗时间
备注：
1. 如果不balance，那么cluster会把新的数据都存放在新的node上，这样会降低mr的工作效率；
2. 也可调用bin/start-balancer.sh 命令执行，也可加参数 -threshold 5
   threshold 是平衡阈值，默认是10%，值越低各节点越平衡，但消耗时间也更长。
3. balancer也可以在有mr job的cluster上运行，默认dfs.balance.bandwidthPerSec很低，为1M/s。在没有mr job时，可以提高该设置加快负载均衡时间。
其他备注：
1. 必须确保slave的firewall已关闭;
2. 确保新的slave的ip已经添加到master及其他slaves的/etc/hosts中，反之也要将master及其他slave的ip添加到新的slave的/etc/hosts中
mapper及reducer个数
url地址： http://wiki.apache.org/hadoop/HowManyMapsAndReduces
HowManyMapsAndReduces
Partitioning your job into maps and reduces
Picking the appropriate size for the tasks for your job can radically change the performance of Hadoop. Increasing the number of tasks increases the framework overhead, but increases load balancing and lowers the cost of failures. At one extreme is the 1 map/1 reduce case where nothing is distributed. The other extreme is to have 1,000,000 maps/ 1,000,000 reduces where the framework runs out of resources for the overhead.
Number of Maps
The number of maps is usually driven by the number of DFS blocks in the input files. Although that causes people to adjust their DFS block size to adjust the number of maps. The right level of parallelism for maps seems to be around 10-100 maps/node, although we have taken it up to 300 or so for very cpu-light map tasks. Task setup takes awhile, so it is best if the maps take at least a minute to execute.
Actually controlling the number of maps is subtle. The mapred.map.tasks parameter is just a hint to the InputFormat for the number of maps. The default InputFormat behavior is to split the total number of bytes into the right number of fragments. However, in the default case the DFS block size of the input files is treated as an upper bound for input splits. A lower bound on the split size can be set via mapred.min.split.size. Thus, if you expect 10TB of input data and have 128MB DFS blocks, you'll end up with 82k maps, unless your mapred.map.tasks is even larger. Ultimately the [WWW] InputFormat determines the number of maps.
The number of map tasks can also be increased manually using the JobConf's conf.setNumMapTasks(int num). This can be used to increase the number of map tasks, but will not set the number below that which Hadoop determines via splitting the input data.
Number of Reduces
The right number of reduces seems to be 0.95 or 1.75 * (nodes * mapred.tasktracker.tasks.maximum). At 0.95 all of the reduces can launch immediately and start transfering map outputs as the maps finish. At 1.75 the faster nodes will finish their first round of reduces and launch a second round of reduces doing a much better job of load balancing.
Currently the number of reduces is limited to roughly 1000 by the buffer size for the output files (io.buffer.size * 2 * numReduces  block size, 会增加块的数量。(猜想从其他节点拿去数据的时候，会合并block，导致block数量增多)
如果设置maximum split size < block size, 会进一步拆分block。
split size = max(minimumSize, min(maximumSize, blockSize));
其中 minimumSize < blockSize < maximumSize.
sort by value
hadoop 不提供直接的sort by value方法，因为这样会降低mapreduce性能。
但可以用组合的办法来实现，具体实现方法见《the definitive guide》, P250
基本思想：
1. 组合key/value作为新的key；
2. 重载partitioner，根据old key来分割；
conf.setPartitionerClass(FirstPartitioner.class);
3. 自定义keyComparator：先根据old key排序，再根据old value排序；
conf.setOutputKeyComparatorClass(KeyComparator.class);
4. 重载GroupComparator, 也根据old key 来组合；  conf.setOutputValueGroupingComparator(GroupComparator.class);
small input files的处理
对于一系列的small files作为input file，会降低hadoop效率。
有3种方法可以将small file合并处理：
1. 将一系列的small files合并成一个sequneceFile，加快mapreduce速度。
详见WholeFileInputFormat及SmallFilesToSequenceFileConverter,《the definitive guide》, P194
2. 使用CombineFileInputFormat集成FileinputFormat，但是未实现过；
3. 使用hadoop archives(类似打包)，减少小文件在namenode中的metadata内存消耗。(这个方法不一定可行，所以不建议使用)
   方法：
   将/my/files目录及其子目录归档成files.har，然后放在/my目录下
   bin/hadoop archive -archiveName files.har /my/files /my
   
   查看files in the archive:
   bin/hadoop fs -lsr har://my/files.har
skip bad records
JobConf conf = new JobConf(ProductMR.class);
conf.setJobName("ProductMR");
conf.setOutputKeyClass(Text.class);
conf.setOutputValueClass(Product.class);
conf.setMapperClass(Map.class);
conf.setReducerClass(Reduce.class);
conf.setMapOutputCompressorClass(DefaultCodec.class);
conf.setInputFormat(SequenceFileInputFormat.class);
conf.setOutputFormat(SequenceFileOutputFormat.class);
String objpath = "abc1";
SequenceFileInputFormat.addInputPath(conf, new Path(objpath));
SkipBadRecords.setMapperMaxSkipRecords(conf, Long.MAX_VALUE);
SkipBadRecords.setAttemptsToStartSkipping(conf, 0);
SkipBadRecords.setSkipOutputPath(conf, new Path("data/product/skip/"));
String output = "abc";
SequenceFileOutputFormat.setOutputPath(conf, new Path(output));
JobClient.runJob(conf);
For skipping failed tasks try : mapred.max.map.failures.percent
restart 单个datanode
如果一个datanode 出现问题，解决之后需要重新加入cluster而不重启cluster，方法如下：
bin/hadoop-daemon.sh start datanode
bin/hadoop-daemon.sh start jobtracker
reduce exceed 100%
"Reduce Task Progress shows > 100% when the total size of map outputs (for a
single reducer) is high "
造成原因：
在reduce的merge过程中，check progress有误差，导致status > 100%，在统计过程中就会出现以下错误：java.lang.ArrayIndexOutOfBoundsException: 3
        at org.apache.hadoop.mapred.StatusHttpServer$TaskGraphServlet.getReduceAvarageProgresses(StatusHttpServer.java:228)
        at org.apache.hadoop.mapred.StatusHttpServer$TaskGraphServlet.doGet(StatusHttpServer.java:159)
        at javax.servlet.http.HttpServlet.service(HttpServlet.java:689)
        at javax.servlet.http.HttpServlet.service(HttpServlet.java:802)
        at org.mortbay.jetty.servlet.ServletHolder.handle(ServletHolder.java:427)
        at org.mortbay.jetty.servlet.WebApplicationHandler.dispatch(WebApplicationHandler.java:475)
        at org.mortbay.jetty.servlet.ServletHandler.handle(ServletHandler.java:567)
        at org.mortbay.http.HttpContext.handle(HttpContext.java:1565)
        at org.mortbay.jetty.servlet.WebApplicationContext.handle(WebApplicationContext.java:635)
        at org.mortbay.http.HttpContext.handle(HttpContext.java:1517)
        at org.mortbay.http.HttpServer.service(HttpServer.java:954)
jira地址：https://issues.apache.org/jira/browse/HADOOP-5210
counters
3中counters：
1. built-in counters: Map input bytes, Map output records...
2. enum counters
   调用方式：
  enum Temperature {
    MISSING,
    MALFORMED
  }
reporter.incrCounter(Temperature.MISSING, 1)
   结果显示：
09/04/20 06:33:36 INFO mapred.JobClient:   Air Temperature Recor
09/04/20 06:33:36 INFO mapred.JobClient:     Malformed=3
09/04/20 06:33:36 INFO mapred.JobClient:     Missing=66136856
3. dynamic countes:
   调用方式：
   reporter.incrCounter("TemperatureQuality", parser.getQuality(),1);
   
   结果显示：
09/04/20 06:33:36 INFO mapred.JobClient:   TemperatureQuality
09/04/20 06:33:36 INFO mapred.JobClient:     2=1246032
09/04/20 06:33:36 INFO mapred.JobClient:     1=973422173
09/04/20 06:33:36 INFO mapred.JobClient:     0=1