Spark Swift 集成 Parquet
Spark Swift Integration Parquet
我使用 Spark 1.3.0 并将 Parquet 文件写入 Openstack Swift 对象存储已有一段时间了。我正在为 parquet 文件使用大约 12 个分区,这些文件将 parquet 文件分成几个部分写入 Swift。写入文件没有问题。但是当我尝试通过 Spark 读取它时,我得到了这个错误:
ERROR Executor: Exception in task 9.0 in stage 2.0 (TID 22)
java.io.EOFException
at java.io.DataInputStream.readFully(DataInputStream.java:197)
at java.io.DataInputStream.readFully(DataInputStream.java:169)
at parquet.hadoop.ParquetFileReader$ConsecutiveChunkList.readAll(ParquetFileReader.java:730)
at parquet.hadoop.ParquetFileReader.readNextRowGroup(ParquetFileReader.java:490)
at parquet.hadoop.InternalParquetRecordReader.checkRead(InternalParquetRecordReader.java:116)
at parquet.hadoop.InternalParquetRecordReader.nextKeyValue(InternalParquetRecordReader.java:193)
at parquet.hadoop.ParquetRecordReader.nextKeyValue(ParquetRecordReader.java:204)
at org.apache.spark.rdd.NewHadoopRDD$$anon.hasNext(NewHadoopRDD.scala:143)
at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:39)
at scala.collection.Iterator$$anon.hasNext(Iterator.scala:327)
at scala.collection.Iterator$$anon.hasNext(Iterator.scala:308)
at scala.collection.Iterator$$anon.hasNext(Iterator.scala:327)
at org.apache.spark.util.collection.ExternalSorter.insertAll(ExternalSorter.scala:210)
at org.apache.spark.shuffle.sort.SortShuffleWriter.write(SortShuffleWriter.scala:63)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:68)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)
at org.apache.spark.scheduler.Task.run(Task.scala:64)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:203)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
我正在使用可通过 https://github.com/openstack/sahara-extra/tree/master/hadoop-swiftfs 获得的 Hadoop-Swift 补丁,它使 Hadoop 能够将 Swift 识别为文件系统。
注意1:如果从Swift下载文件到本地文件系统,Spark可以完美读取文件。
注意2:我也注意到如果不分区parquet文件,读取是完美的。
有人在使用 Spark、Openstack Swift 吗?
我一直在使用带有 Openstack Swift 的 Spark 1.3.0 作为我的存储后端,但遇到了完全相同的问题。
使用 Spark 的缓存功能提供了一个有效的解决方法,等待对 hadoop-cloud 的 swift 支持的实际修复。
val df = spark.read.parquet("your_swift_url").persist()
我使用 Spark 1.3.0 并将 Parquet 文件写入 Openstack Swift 对象存储已有一段时间了。我正在为 parquet 文件使用大约 12 个分区,这些文件将 parquet 文件分成几个部分写入 Swift。写入文件没有问题。但是当我尝试通过 Spark 读取它时,我得到了这个错误:
ERROR Executor: Exception in task 9.0 in stage 2.0 (TID 22)
java.io.EOFException
at java.io.DataInputStream.readFully(DataInputStream.java:197)
at java.io.DataInputStream.readFully(DataInputStream.java:169)
at parquet.hadoop.ParquetFileReader$ConsecutiveChunkList.readAll(ParquetFileReader.java:730)
at parquet.hadoop.ParquetFileReader.readNextRowGroup(ParquetFileReader.java:490)
at parquet.hadoop.InternalParquetRecordReader.checkRead(InternalParquetRecordReader.java:116)
at parquet.hadoop.InternalParquetRecordReader.nextKeyValue(InternalParquetRecordReader.java:193)
at parquet.hadoop.ParquetRecordReader.nextKeyValue(ParquetRecordReader.java:204)
at org.apache.spark.rdd.NewHadoopRDD$$anon.hasNext(NewHadoopRDD.scala:143)
at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:39)
at scala.collection.Iterator$$anon.hasNext(Iterator.scala:327)
at scala.collection.Iterator$$anon.hasNext(Iterator.scala:308)
at scala.collection.Iterator$$anon.hasNext(Iterator.scala:327)
at org.apache.spark.util.collection.ExternalSorter.insertAll(ExternalSorter.scala:210)
at org.apache.spark.shuffle.sort.SortShuffleWriter.write(SortShuffleWriter.scala:63)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:68)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)
at org.apache.spark.scheduler.Task.run(Task.scala:64)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:203)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
我正在使用可通过 https://github.com/openstack/sahara-extra/tree/master/hadoop-swiftfs 获得的 Hadoop-Swift 补丁,它使 Hadoop 能够将 Swift 识别为文件系统。
注意1:如果从Swift下载文件到本地文件系统,Spark可以完美读取文件。
注意2:我也注意到如果不分区parquet文件,读取是完美的。
有人在使用 Spark、Openstack Swift 吗?
我一直在使用带有 Openstack Swift 的 Spark 1.3.0 作为我的存储后端,但遇到了完全相同的问题。
使用 Spark 的缓存功能提供了一个有效的解决方法,等待对 hadoop-cloud 的 swift 支持的实际修复。
val df = spark.read.parquet("your_swift_url").persist()