前言
近日在线上发现有些mapreduce作业的执行时间很长,我们需要解决这个问题。输入文件的大小是5G,采用了lzo压缩,整个集群的默认block大小是128M。本文将详细描述这次线上问题的排查过程。
现象
线上有一个脚本,为了便于展示,我将这个脚本重新copy了一份并重命名为zzz。这个脚本实际是使用Hadoop streaming运行一个mapreduce任务,在线上执行它的部分输出内容如下:
可以看到map任务划分为1个。这个执行过程十分漫长,我将中间的一些信息省略,map与reduce任务的执行进度如下:
16/05/16 10:22:16 INFO mapreduce.Job: map 0% reduce 0%
16/05/16 10:22:32 INFO mapreduce.Job: map 1% reduce 0%
。。。
16/05/16 10:44:14 INFO mapreduce.Job: map 99% reduce 0%
16/05/16 10:44:20 INFO mapreduce.Job: map 100% reduce 0%
16/05/16 10:44:33 INFO mapreduce.Job: map 100% reduce 2%
16/05/16 10:44:34 INFO mapreduce.Job: map 100% reduce 19%
。。。
16/05/16 10:44:57 INFO mapreduce.Job: map 100% reduce 99%
16/05/16 10:44:58 INFO mapreduce.Job: map 100% reduce 100%
从以上内容可以看到map任务执行一共耗时22分钟左右,而reduce任务只耗用了30多秒。
分析
根据以上现象分析,我们知道耗时主要发生在map任务执行的阶段。我们首先查看下这个map任务的输入内容,看到它的大小为5GB且使用lzo压缩:
如此大的输入仅仅在一个map任务中处理显然是进度缓慢的主要原因,我们需要对mapreduce的任务划分进行干预。我们查看下mapreduce任务的InputFormat,以便确定干预的手段,打开我们的脚本其中有以下代码片段:
hadoop jar $streaming_jar \
-D mapred.reduce.tasks=30 \
-D mapreduce.job.maps=100 \
-D mapreduce.input.fileinputformat.split.minsize=100000000 \
-inputformat TextInputFormat \
-file mr.py \
-input $member_input \
-output $output \
-mapper "python mr.py map" \
-reducer "python mr.py reduce"
我们看到甚至在脚本中还配置了mapreduce.job.maps,根据《Hadoop2.6.0的FileInputFormat的任务切分原理分析(即如何控制FileInputFormat的map任务数量)》一文的分析,我们知道此参数实际不会对map任务划分产生任何影响。查看到mapreduce作业的input format是TextInputFormat,TextInputFormat的实现见代码清单1。
代码清单1 TextInputFormat的实现
/** An {@link InputFormat} for plain text files. Files are broken into lines.
* Either linefeed or carriage-return are used to signal end of line. Keys are
* the position in the file, and values are the line of text.. */
@InterfaceAudience.Public
@InterfaceStability.Stable
public class TextInputFormat extends FileInputFormat<LongWritable, Text> {
//省略无关代码
@Override
protected boolean isSplitable(JobContext context, Path file) {
final CompressionCodec codec =
new CompressionCodecFactory(context.getConfiguration()).getCodec(file);
if (null == codec) {
return true;
}
return codec instanceof SplittableCompressionCodec;
}
}
我们看到TextInputFormat继承了FileInputFormat,因而根据《Hadoop2.6.0的FileInputFormat的任务切分原理分析(即如何控制FileInputFormat的map任务数量)》一文的内容,真正影响使用FileInputFormat的map任务划分的参数有:
- dfs.blockSize
- mapreduce.input.fileinputformat.split.minsize
- mapreduce.input.fileinputformat.split.maxsize
<property>
<name>io.compression.codecs</name>
<value>org.apache.hadoop.io.compress.GzipCodec,org.apache.hadoop.io.compress.DefaultCodec,org.apache.hadoop.io.compress.BZip2Codec,com.hadoop.compression.lzo.LzoCodec,com.hadoop.compression.lzo.LzopCodec,org.apache.hadoop.io.compress.SnappyCodec</value>
<source>core-site.xml</source>
</property>
/**
* A {@link org.apache.hadoop.io.compress.CompressionCodec} for a streaming
* <b>lzo</b> compression/decompression pair.
* http://www.oberhumer.com/opensource/lzo/
*
*/
public class LzoCodec extends org.anarres.lzo.hadoop.codec.LzoCodec {
}
/**
* A {@link org.apache.hadoop.io.compress.CompressionCodec} for a streaming
* <b>lzo</b> compression/decompression pair.
* http://www.oberhumer.com/opensource/lzo/
*
*/
public class LzoCodec extends Configured implements CompressionCodec {
可以看到org.anarres.lzo.hadoop.codec.LzoCodec直接实现了CompressionCodec,并没有实现SplittableCompressionCodec接口,SplittableCompressionCodec接口实际也继承了CompressionCodec接口:
@InterfaceAudience.Public
@InterfaceStability.Evolving
public interface SplittableCompressionCodec extends CompressionCodec {
因此codec instanceof SplittableCompressionCodec这条Java语句将返回false,采用lzo压缩算法的输入文件将导致map任务不可划分,也就是不会生成多个map任务。
解决方法
后记:个人总结整理的《深入理解Spark:核心思想与源码分析》一书现在已经正式出版上市,目前京东、当当、天猫等网站均有销售,欢迎感兴趣的同学购买。
京东:http://item.jd.com/11846120.html
当当:http://product.dangdang.com/23838168.html