线上Spark处理Bzip2引出Hadoop Bzip2线程安全有关问题
我们的Hadoop生产环境有两个版本，其中一个是1.0.3，为了支持日志压缩和split，我们添加了hadoop-1.2中关于Bzip2压缩的feature. 一切运行良好。
为了满足公司对迭代计算的需求(复杂HiveSQL,广告推荐算法，机器学习 etc), 我们构建了自己的Spark集群，最初是Standalone Mode，版本spark-0.9.1,支持Shark。
上线后，问题接踵而来，最为致命的是，shark在处理Hadooop bzip2文件时计算结果通常会有偏差,有时差的特别离谱(比如，用shark统计1个5kw行的日志,结果只有
3kw行).
显然shark+hive+spark+hadoop的某个环节出了bug。第一次面对这么复杂的系统，着实头疼。
于是，开始蛮干，部署shark+hive+spark+hadoop开发环境，debug，查看出问题的环节。（这个过程中把Spark-core的源码也缕了一遍),始终没有发现什么问题。
后来，参加了Spark技术大会，和同行交流的过程中，幡然悔悟： Spark的task是线程级并发的，而Hadoop MR的task是进程级并发的，那么，会不会是Bzip2存在线程安全问题呢？
回来后，查看Bzip2Codec相关的代码，终于发现了问题所在。(话说，凌晨3点，没有eclipse,用vim 改的), 迫不及待的重新编译Hadoop和Spark，测试，发现处理Bzip2结果OK了！
CBzip2InputStream 有一个flag变量private static boolean skipDecompression ,默认值是false. (表示是否要跳出解压缩？)
这个变量在CBZip2InputStream的 public static long numberOfBytesTillNextMarker方法里被置为true. 在CBZip2InputStream.read方法里可能被置为false；
看来，它是一个流(inputStream)读取过程中经常用到的flag变量.
假设，现在同一个进程内，有两个线程都要做Bzip2流的读取工作，它们共用一个static类型的标志位skipDecompression. 显然会出问题的。
解决办法：将其skipDecompression改成非static类型，修改调用skipDecompression的静态方法为非static类型。
(PS: Bzip2Codec.createInputStream 方法用于将给定的流InputStream转换成CBzip2InputStream. 是Bizp2文件读取时的必经方法。该方法里调用了CBZip2InputStream.numberOfBytesTillNextMarker静态方法，这里导致了多线程读取BZip2流时skipDecompresion标志位混乱)
由于向社区提交path需要漫长的过程，暂时没有提交社区。具体的patch如下，如有同行遇到同类问题，请借鉴.
Index: src/core/org/apache/hadoop/io/compress/bzip2/CBZip2InputStream.java
===================================================================
--- src/core/org/apache/hadoop/io/compress/bzip2/CBZip2InputStream.java (版本 510)
+++ src/core/org/apache/hadoop/io/compress/bzip2/CBZip2InputStream.java (版本 525)
@@ -129,7 +129,9 @@
private int computedBlockCRC, computedCombinedCRC;
private boolean skipResult = false;// used by skipToNextMarker
- private static boolean skipDecompression = false;
+ //modified by jicheng.song
+ //private static boolean skipDecompression = false;
+ private boolean skipDecompression = false;
// Variables used by setup* methods exclusively
@@ -315,13 +317,18 @@
* @throws IOException
*
*/
- public static long numberOfBytesTillNextMarker(final InputStream in) throws IOException{
- CBZip2InputStream.skipDecompression = true;
- CBZip2InputStream anObject = null;
+ //modified by jicheng.song
+ //public static long numberOfBytesTillNextMarker(final InputStream in) throws IOException{
+ public long numberOfBytesTillNextMarker(final InputStream in) throws IOException{
+ this.skipDecompression = true;
+ //
+ this.in = new BufferedInputStream(in, 1024 * 9);// >1 MB buffer
+ this.readMode = readMode;
+ //CBZip2InputStream anObject = null;
- anObject = new CBZip2InputStream(in, READ_MODE.BYBLOCK);
+ //anObject = new CBZip2InputStream(in, READ_MODE.BYBLOCK);
- return anObject.getProcessedByteCount();
+ return this.getProcessedByteCount();
}
public CBZip2InputStream(final InputStream in) throws IOException {
@@ -395,7 +402,9 @@
if(skipDecompression){
changeStateToProcessABlock();
- CBZip2InputStream.skipDecompression = false;
+ //modified by jicheng.song
+ //CBZip2InputStream.skipDecompression = false;
+ this.skipDecompression = false;
}
final int hi = offs + len;
Index: src/core/org/apache/hadoop/io/compress/BZip2Codec.java
===================================================================
--- src/core/org/apache/hadoop/io/compress/BZip2Codec.java (版本 510)
+++ src/core/org/apache/hadoop/io/compress/BZip2Codec.java (版本 525)
@@ -148,8 +148,11 @@
// BZip2 start of block markers are of 6 bytes. But the very first block
// also has "BZh9", making it 10 bytes. This is the common case. But at
// time stream might start without a leading BZ.
+ CBZip2InputStream tmpInput = new CBZip2InputStream(seekableIn, readMode);
final long FIRST_BZIP2_BLOCK_MARKER_POSITION =
- CBZip2InputStream.numberOfBytesTillNextMarker(seekableIn);
+ //modified by jicheng.song
+ //CBZip2InputStream.numberOfBytesTillNextMarker(seekableIn);
+ tmpInput.numberOfBytesTillNextMarker(seekableIn);
long adjStart = Math.max(0L, start - FIRST_BZIP2_BLOCK_MARKER_POSITION);
((Seekable)seekableIn).seek(adjStart);