Spark com.databricks.spark.csv无法使用node-snappy加载snappy压缩文件

我在S3上有一些使用snappy压缩算法压缩的csv文件（使用 node-snappy 包） . 我喜欢使用 com.databricks.spark.csv 在spark中处理这些文件，但我一直收到无效的文件输入错误 .

码：

file_df = sqlContext.read.format('com.databricks.spark.csv').options(header='true', inferschema='true', codec='snappy', mode='FAILFAST').load('s3://sample.csv.snappy')

错误信息：

16/09/24 21:57:25 WARN TaskSetManager：阶段0.0中丢失的任务0.0（TID 0，ip-10-0-32-5.ec2.internal）：java.lang.InternalError：无法解压缩数据 . 输入无效 . org.apache.hadoop.io.compress.snappy.SnappyDecompressor.decompressBytesDirect（Native Method）位于org.apache.hadoop的org.apache.hadoop.io.compress.snappy.SnappyDecompressor.decompress(SnappyDecompressor.java:239） . io.compress.BlockDecompressorStream.decompress（BlockDecompressorStream.java:88）atg.apache.hadoop.io.compress.DecompressorStream.read（DecompressorStream.java:85）at java.io.InputStream.read（InputStream.java:101）在org.apache.hadoop.util.LineReader.fillBuffer（LineReader.java:180）在org.apache.hadoop.util.LineReader.readDefaultLine（LineReader.java:216）在org.apache.hadoop.util.LineReader.readLine （LineReader.java:174）在org.apache.hadoop.mapred.LineRecordReader.skipUtfByteOrderMark（LineRecordReader.java:208）在org.apache.hadoop.mapred.LineRecordReader.next（LineRecordReader.java:246）在org.apache . hadoop.mapred.LineRecordReader.next（LineRecordReader.java:48）atg.apache.spark.rdd.HadoopRDD $$ anon $ 1.getNext（HadoopRDD.scala：255）at org.apache.spark.rdd.HadoopRDD $$ anon $ 1位于org.apache.spark.util.NextIterator.hasNext（NextIterator.scala：73）的getNext（HadoopRDD.scala：209）位于scala.collection.Iterator的org.apache.spark.InterruptibleIterator.hasNext（InterruptibleIterator.scala：39） $ anon $ 11.hasNext（Iterator.scala：408）at scala.collection.Iterator $$ anon $ 13.hasNext（Iterator.scala：461）at scala.collection.Iterator $$ anon $ 10.hasNext（Iterator.scala：389） ）scala.collection.Iterator $ class.foreach（Iterator.scala：893）at scala.collection.AbstractIterator.foreach（Iterator.scala：1336）at scala.collection.generic.Growable $ class . $ plus $ plus $ eq （Growable.scala：59）at scala.collection.mutable.ArrayBuffer . $ plus $ plus $ eq（ArrayBuffer.scala：104）at scala.collection.mutable.ArrayBuffer . $ plus $ plus $ eq（ArrayBuffer.scala：48 ）scala.collection.TraversableOnce $ class.to（TraversableOnce.scala：310）at scala.collection.AbstractIterator.to（Iterator.scala：1336）at scala.collection.TraversableOnce $ class.toBuffer（TraversableOnce.scala：302）在scala.collection.AbstractIterator.toBuffer（Iterato r.scala：1336）scala.collection.TraversableOnce $ class.toArray（TraversableOnce.scala：289）at scala.collection.AbstractIterator.toArray（Iterator.scala：1336）at org.apache.spark.rdd.RDD $$ anonfun $在org.apache上获取$ 1 $$ anonfun $ 29.apply（RDD.scala：1305）org.apache.spark.rdd.RDD $$ anonfun $ take $ 1 $$ anonfun $ 29.apply（RDD.scala：1305）at org.apache .spark.SparkContext $$ anonfun $ runJob $ 5.apply（SparkContext.scala：1897）org.apache.spark.SparkContext $$ anonfun $ runJob $ 5.apply（SparkContext.scala：1897）at org.apache.spark.scheduler .ResultTask.runTask（ResultTask.scala：70）atg.apache.spark.scheduler.Task.run（Task.scala：85）at org.apache.spark.executor.Executor $ TaskRunner.run（Executor.scala：274） ）java.util.concurrent.ThreadPoolExecutor.runWorker（ThreadPoolExecutor.java:1142）at java.util.concurrent.ThreadPoolExecutor $ Worker.run（ThreadPoolExecutor.java:617）at java.lang.Thread.run（Thread.java： 745）