【发布时间】：2017-02-02 12:53:52
【问题描述】：
我在 S3 上有一些使用 snappy 压缩算法（使用 node-snappy 包）压缩的 csv 文件。我喜欢使用 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（本机方法） 在 org.apache.hadoop.io.compress.snappy.SnappyDecompressor.decompress(SnappyDecompressor.java:239) 在 org.apache.hadoop.io.compress.BlockDecompressorStream.decompress(BlockDecompressorStream.java:88) 在 org.apache.hadoop.io.compress.DecompressorStream.read(DecompressorStream.java:85) 在 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) 在 org.apache.spark.rdd.HadoopRDD$$anon$1.getNext(HadoopRDD.scala:255) 在 org.apache.spark.rdd.HadoopRDD$$anon$1.getNext(HadoopRDD.scala:209) 在 org.apache.spark.util.NextIterator.hasNext(NextIterator.scala:73) 在 org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:39) 在 scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408) 在 scala.collection.Iterator$$anon$13.hasNext(Iterator.scala:461) 在 scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:389) 在 scala.collection.Iterator$class.foreach(Iterator.scala:893) 在 scala.collection.AbstractIterator.foreach(Iterator.scala:1336) 在 scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:59) 在 scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:104) 在 scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:48) 在 scala.collection.TraversableOnce$class.to(TraversableOnce.scala:310) 在 scala.collection.AbstractIterator.to(Iterator.scala:1336) 在 scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:302) 在 scala.collection.AbstractIterator.toBuffer(Iterator.scala:1336) 在 scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:289) 在 scala.collection.AbstractIterator.toArray(Iterator.scala:1336) 在 org.apache.spark.rdd.RDD$$anonfun$take$1$$anonfun$29.apply(RDD.scala:1305) 在 org.apache.spark.rdd.RDD$$anonfun$take$1$$anonfun$29.apply(RDD.scala:1305) 在 org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1897) 在 org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1897) 在 org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:70) 在 org.apache.spark.scheduler.Task.run(Task.scala:85) 在 org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:274) 在 java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) 在 java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) 在 java.lang.Thread.run(Thread.java:745)
【问题讨论】：
-
如果你使用的是 spark 2.0，试试这个类的 csv 方法：
df = spark.read.csv('csv.file') -
spark.read.csv 使用
com.databricks.spark.csv，同样的问题仍然存在。
标签： apache-spark pyspark snappy databricks apache-spark-2.0