Hi Andrew, I submitted a patch and verified it solves the problem. You can download the patch from https://issues.apache.org/jira/browse/HADOOP-10614 .
Best, Xiangrui On Fri, May 16, 2014 at 6:48 PM, Xiangrui Meng <men...@gmail.com> wrote: > Hi Andrew, > > This is the JIRA I created: > https://issues.apache.org/jira/browse/MAPREDUCE-5893 . Hopefully > someone wants to work on it. > > Best, > Xiangrui > > On Fri, May 16, 2014 at 6:47 PM, Xiangrui Meng <men...@gmail.com> wrote: >> Hi Andre, >> >> I could reproduce the bug with Hadoop 2.2.0. Some older version of >> Hadoop do not support splittable compression, so you ended up with >> sequential reads. It is easy to reproduce the bug with the following >> setup: >> >> 1) Workers are configured with multiple cores. >> 2) BZip2 files are big enough or minPartitions is large enough when >> you load the file via sc.textFile(), so that one worker has more than >> one tasks. >> >> Best, >> Xiangrui >> >> On Fri, May 16, 2014 at 4:06 PM, Andrew Ash <and...@andrewash.com> wrote: >>> Hi Xiangrui, >>> >>> // FYI I'm getting your emails late due to the Apache mailing list outage >>> >>> I'm using CDH4.4.0, which I think uses the MapReduce v2 API. The .jars are >>> named like this: hadoop-hdfs-2.0.0-cdh4.4.0.jar >>> >>> I'm also glad you were able to reproduce! Please paste a link to the Hadoop >>> bug you file so I can follow along. >>> >>> Thanks! >>> Andrew >>> >>> >>> On Tue, May 13, 2014 at 9:08 AM, Xiangrui Meng <men...@gmail.com> wrote: >>>> >>>> Which hadoop version did you use? I'm not sure whether Hadoop v2 fixes >>>> the problem you described, but it does contain several fixes to bzip2 >>>> format. -Xiangrui >>>> >>>> On Wed, May 7, 2014 at 9:19 PM, Andrew Ash <and...@andrewash.com> wrote: >>>> > Hi all, >>>> > >>>> > Is anyone reading and writing to .bz2 files stored in HDFS from Spark >>>> > with >>>> > success? >>>> > >>>> > >>>> > I'm finding the following results on a recent commit (756c96 from 24hr >>>> > ago) >>>> > and CDH 4.4.0: >>>> > >>>> > Works: val r = sc.textFile("/user/aa/myfile.bz2").count >>>> > Doesn't work: val r = sc.textFile("/user/aa/myfile.bz2").map((s:String) >>>> > => >>>> > s+"| " ).count >>>> > >>>> > Specifically, I'm getting an exception coming out of the bzip2 libraries >>>> > (see below stacktraces), which is unusual because I'm able to read from >>>> > that >>>> > file without an issue using the same libraries via Pig. It was >>>> > originally >>>> > created from Pig as well. >>>> > >>>> > Digging a little deeper I found this line in the .bz2 decompressor's >>>> > javadoc >>>> > for CBZip2InputStream: >>>> > >>>> > "Instances of this class are not threadsafe." [source] >>>> > >>>> > >>>> > My current working theory is that Spark has a much higher level of >>>> > parallelism than Pig/Hadoop does and thus I get these wild >>>> > IndexOutOfBounds >>>> > exceptions much more frequently (as in can't finish a run over a little >>>> > 2M >>>> > row file) vs hardly at all in other libraries. >>>> > >>>> > The only other reference I could find to the issue was in presto-users, >>>> > but >>>> > the recommendation to leave .bz2 for .lzo doesn't help if I actually do >>>> > want >>>> > the higher compression levels of .bz2. >>>> > >>>> > >>>> > Would love to hear if I have some kind of configuration issue or if >>>> > there's >>>> > a bug in .bz2 that's fixed in later versions of CDH, or generally any >>>> > other >>>> > thoughts on the issue. >>>> > >>>> > >>>> > Thanks! >>>> > Andrew >>>> > >>>> > >>>> > >>>> > Below are examples of some exceptions I'm getting: >>>> > >>>> > 14/05/07 15:09:49 WARN scheduler.TaskSetManager: Loss was due to >>>> > java.lang.ArrayIndexOutOfBoundsException >>>> > java.lang.ArrayIndexOutOfBoundsException: 65535 >>>> > at >>>> > >>>> > org.apache.hadoop.io.compress.bzip2.CBZip2InputStream.hbCreateDecodeTables(CBZip2InputStream.java:663) >>>> > at >>>> > >>>> > org.apache.hadoop.io.compress.bzip2.CBZip2InputStream.createHuffmanDecodingTables(CBZip2InputStream.java:790) >>>> > at >>>> > >>>> > org.apache.hadoop.io.compress.bzip2.CBZip2InputStream.recvDecodingTables(CBZip2InputStream.java:762) >>>> > at >>>> > >>>> > org.apache.hadoop.io.compress.bzip2.CBZip2InputStream.getAndMoveToFrontDecode(CBZip2InputStream.java:798) >>>> > at >>>> > >>>> > org.apache.hadoop.io.compress.bzip2.CBZip2InputStream.initBlock(CBZip2InputStream.java:502) >>>> > at >>>> > >>>> > org.apache.hadoop.io.compress.bzip2.CBZip2InputStream.changeStateToProcessABlock(CBZip2InputStream.java:333) >>>> > at >>>> > >>>> > org.apache.hadoop.io.compress.bzip2.CBZip2InputStream.read(CBZip2InputStream.java:397) >>>> > at >>>> > >>>> > org.apache.hadoop.io.compress.BZip2Codec$BZip2CompressionInputStream.read(BZip2Codec.java:426) >>>> > at java.io.InputStream.read(InputStream.java:101) >>>> > at >>>> > org.apache.hadoop.util.LineReader.readDefaultLine(LineReader.java:209) >>>> > at >>>> > org.apache.hadoop.util.LineReader.readLine(LineReader.java:173) >>>> > at >>>> > >>>> > org.apache.hadoop.mapred.LineRecordReader.next(LineRecordReader.java:203) >>>> > at >>>> > org.apache.hadoop.mapred.LineRecordReader.next(LineRecordReader.java:43) >>>> > >>>> > >>>> > >>>> > >>>> > java.lang.ArrayIndexOutOfBoundsException: 900000 >>>> > at >>>> > >>>> > org.apache.hadoop.io.compress.bzip2.CBZip2InputStream.getAndMoveToFrontDecode(CBZip2InputStream.java:900) >>>> > at >>>> > >>>> > org.apache.hadoop.io.compress.bzip2.CBZip2InputStream.initBlock(CBZip2InputStream.java:502) >>>> > at >>>> > >>>> > org.apache.hadoop.io.compress.bzip2.CBZip2InputStream.changeStateToProcessABlock(CBZip2InputStream.java:333) >>>> > at >>>> > >>>> > org.apache.hadoop.io.compress.bzip2.CBZip2InputStream.read(CBZip2InputStream.java:397) >>>> > at >>>> > >>>> > org.apache.hadoop.io.compress.BZip2Codec$BZip2CompressionInputStream.read(BZip2Codec.java:426) >>>> > at java.io.InputStream.read(InputStream.java:101) >>>> > at >>>> > org.apache.hadoop.util.LineReader.readDefaultLine(LineReader.java:209) >>>> > at >>>> > org.apache.hadoop.util.LineReader.readLine(LineReader.java:173) >>>> > at >>>> > >>>> > org.apache.hadoop.mapred.LineRecordReader.next(LineRecordReader.java:203) >>>> > at >>>> > org.apache.hadoop.mapred.LineRecordReader.next(LineRecordReader.java:43) >>>> > at >>>> > org.apache.spark.rdd.HadoopRDD$$anon$1.getNext(HadoopRDD.scala:198) >>>> > at >>>> > org.apache.spark.rdd.HadoopRDD$$anon$1.getNext(HadoopRDD.scala:181) >>>> > at >>>> > org.apache.spark.util.NextIterator.hasNext(NextIterator.scala:71) >>>> > at >>>> > >>>> > org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:35) >>>> > at >>>> > scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327) >>>> > at >>>> > scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327) >>>> > at >>>> > scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327) >>>> > at scala.collection.Iterator$class.foreach(Iterator.scala:727) >>>> > at >>>> > scala.collection.AbstractIterator.foreach(Iterator.scala:1157) >>>> > at >>>> > >>>> > org.apache.spark.rdd.RDD.org$apache$spark$rdd$RDD$$countPartition$1(RDD.scala:868) >>>> > >>>> > >>>> > >>>> > java.lang.ArrayIndexOutOfBoundsException: -921878509 >>>> > at >>>> > >>>> > org.apache.hadoop.io.compress.bzip2.CBZip2InputStream.getAndMoveToFrontDecode0(CBZip2InputStream.java:1011) >>>> > at >>>> > >>>> > org.apache.hadoop.io.compress.bzip2.CBZip2InputStream.getAndMoveToFrontDecode(CBZip2InputStream.java:826) >>>> > at >>>> > >>>> > org.apache.hadoop.io.compress.bzip2.CBZip2InputStream.initBlock(CBZip2InputStream.java:502) >>>> > at >>>> > >>>> > org.apache.hadoop.io.compress.bzip2.CBZip2InputStream.changeStateToProcessABlock(CBZip2InputStream.java:333) >>>> > at >>>> > >>>> > org.apache.hadoop.io.compress.bzip2.CBZip2InputStream.read(CBZip2InputStream.java:397) >>>> > at >>>> > >>>> > org.apache.hadoop.io.compress.BZip2Codec$BZip2CompressionInputStream.read(BZip2Codec.java:432) >>>> > at java.io.InputStream.read(InputStream.java:101) >>>> > at >>>> > org.apache.hadoop.util.LineReader.readDefaultLine(LineReader.java:209) >>>> > at >>>> > org.apache.hadoop.util.LineReader.readLine(LineReader.java:173) >>>> > at >>>> > >>>> > org.apache.hadoop.mapred.LineRecordReader.next(LineRecordReader.java:203) >>>> > at >>>> > org.apache.hadoop.mapred.LineRecordReader.next(LineRecordReader.java:43) >>>> > at >>>> > org.apache.spark.rdd.HadoopRDD$$anon$1.getNext(HadoopRDD.scala:198) >>>> > at >>>> > org.apache.spark.rdd.HadoopRDD$$anon$1.getNext(HadoopRDD.scala:181) >>>> > at >>>> > org.apache.spark.util.NextIterator.hasNext(NextIterator.scala:71) >>>> > at >>>> > >>>> > org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:35) >>>> > at >>>> > scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327) >>>> > at >>>> > scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327) >>>> > at >>>> > scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327) >>>> > at scala.collection.Iterator$class.foreach(Iterator.scala:727) >>>> > at >>>> > scala.collection.AbstractIterator.foreach(Iterator.scala:1157) >>>> > at >>>> > >>>> > org.apache.spark.rdd.RDD.org$apache$spark$rdd$RDD$$countPartition$1(RDD.scala:868) >>>> > at org.apache.spark.rdd.RDD$$anonfun$24.apply(RDD.scala:879) >>>> > at org.apache.spark.rdd.RDD$$anonfun$24.apply(RDD.scala:879) >>>> > at org.apache.spark.rdd.RDD$$anonfun$12.apply(RDD.scala:548) >>>> > at org.apache.spark.rdd.RDD$$anonfun$12.apply(RDD.scala:548) >>>> > >>>> > >>>> > >>>> > java.lang.ArrayIndexOutOfBoundsException: -1321104434 >>>> > at >>>> > >>>> > org.apache.hadoop.io.compress.bzip2.CBZip2InputStream.getAndMoveToFrontDecode0(CBZip2InputStream.java:1011) >>>> > at >>>> > >>>> > org.apache.hadoop.io.compress.bzip2.CBZip2InputStream.getAndMoveToFrontDecode(CBZip2InputStream.java:826) >>>> > at >>>> > >>>> > org.apache.hadoop.io.compress.bzip2.CBZip2InputStream.initBlock(CBZip2InputStream.java:502) >>>> > at >>>> > >>>> > org.apache.hadoop.io.compress.bzip2.CBZip2InputStream.changeStateToProcessABlock(CBZip2InputStream.java:333) >>>> > at >>>> > >>>> > org.apache.hadoop.io.compress.bzip2.CBZip2InputStream.read(CBZip2InputStream.java:397) >>>> > at >>>> > >>>> > org.apache.hadoop.io.compress.BZip2Codec$BZip2CompressionInputStream.read(BZip2Codec.java:426) >>>> > at java.io.InputStream.read(InputStream.java:101) >>>> > at >>>> > org.apache.hadoop.util.LineReader.readDefaultLine(LineReader.java:209) >>>> > at >>>> > org.apache.hadoop.util.LineReader.readLine(LineReader.java:173) >>>> > at >>>> > >>>> > org.apache.hadoop.mapred.LineRecordReader.next(LineRecordReader.java:203) >>>> > at >>>> > org.apache.hadoop.mapred.LineRecordReader.next(LineRecordReader.java:43) >>>> > at >>>> > org.apache.spark.rdd.HadoopRDD$$anon$1.getNext(HadoopRDD.scala:198) >>>> > at >>>> > org.apache.spark.rdd.HadoopRDD$$anon$1.getNext(HadoopRDD.scala:181) >>>> > at >>>> > org.apache.spark.util.NextIterator.hasNext(NextIterator.scala:71) >>>> > at >>>> > >>>> > org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:35) >>>> > at >>>> > scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327) >>>> > at >>>> > scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327) >>>> > at >>>> > scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327) >>>> > at scala.collection.Iterator$class.foreach(Iterator.scala:727) >>>> > at >>>> > scala.collection.AbstractIterator.foreach(Iterator.scala:1157) >>>> > at >>>> > >>>> > org.apache.spark.rdd.RDD.org$apache$spark$rdd$RDD$$countPartition$1(RDD.scala:868) >>> >>>
