Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/1658#discussion_r18612406
--- Diff: core/src/test/scala/org/apache/spark/FileSuite.scala ---
@@ -224,6 +226,128 @@ class FileSuite extends FunSuite with
LocalSparkContext {
assert(output.map(_.toString).collect().toList === List("(1,a)",
"(2,aa)", "(3,aaa)"))
}
+ test("binary file input as byte array") {
+ sc = new SparkContext("local", "test")
+ val outFile = new File(tempDir, "record-bytestream-00000.bin")
+ val outFileName = outFile.getAbsolutePath()
+
+ // create file
+ val testOutput = Array[Byte](1,2,3,4,5,6)
+ val bbuf = java.nio.ByteBuffer.wrap(testOutput)
+ // write data to file
+ val file = new java.io.FileOutputStream(outFile)
+ val channel = file.getChannel
+ channel.write(bbuf)
+ channel.close()
+ file.close()
+
+ val inRdd = sc.binaryFiles(outFileName)
+ val (infile: String, indata: PortableDataStream) = inRdd.first
+
+ // Make sure the name and array match
+ assert(infile.contains(outFileName)) // a prefix may get added
+ assert(indata.toArray === testOutput)
+ }
+
+ test("portabledatastream caching tests") {
+ sc = new SparkContext("local", "test")
+ val outFile = new File(tempDir, "record-bytestream-00000.bin")
+ val outFileName = outFile.getAbsolutePath()
+
+ // create file
+ val testOutput = Array[Byte](1,2,3,4,5,6)
+ val bbuf = java.nio.ByteBuffer.wrap(testOutput)
+ // write data to file
+ val file = new java.io.FileOutputStream(outFile)
+ val channel = file.getChannel
+ channel.write(bbuf)
+ channel.close()
+ file.close()
+
+ val inRdd = sc.binaryFiles(outFileName).cache()
+ inRdd.foreach{
+ curData: (String, PortableDataStream) =>
+ curData._2.toArray() // force the file to read
+ }
+ val mappedRdd = inRdd.map{
+ curData: (String, PortableDataStream) =>
+ (curData._2.getPath(),curData._2)
+ }
+ val (infile: String, indata: PortableDataStream) = mappedRdd.first
+
+ // Try reading the output back as an object file
+
+ assert(indata.toArray === testOutput)
+ }
--- End diff --
Apart from the cache() test, try adding one where we call
persist(StorageLevel.DISK_ONLY) to check that these are also stored correctly
to disk. And add one where we use Kryo serialization too.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
