How to save a BufferedImage RDD as HDFS file

Question

I have a requirement to read Image from HDFS, do some processing and save the image back to HDFS. This processing has to be done in spark. I am reading the images files as sc.binaryFiles and then converting them into Buffered Images and performing some operations. But I get "Task not serializable" error when I try to save an RDD[BufferedImage] to a FSDataOutputStream

    //read binary files from RDD
    val images = sc.binaryFiles("/tmp/images/")
    //images: org.apache.spark.rdd.RDD[(String, org.apache.spark.input.PortableDataStream)] = /tmp/images/ 

    //get BufferedImageRDD
    val bufImages = images.map(x => ImageIO.read(x._2.open))
    //bufImages: org.apache.spark.rdd.RDD[java.awt.image.BufferedImage] = MapPartitionsRDD[1]


    //try saving in local directory
    bufImages.foreach(x => UtilImageIO.saveImage(x,"Mean3.jpg"))
    //success



    //try saving in hdfs

    val conf = new Configuration()
    val fileSystem = FileSystem.get(conf);
    val out = fileSystem.create(new Path("/tmp/img1.png"));
    //out: org.apache.hadoop.fs.FSDataOutputStream = org.apache.hadoop.hdfs.client.HdfsDataOutputStream@440f55ad


    bufImages.foreach(x => ImageIO.write(x,"png", out))

The above code throws following error

    org.apache.spark.SparkException: Task not serializable
      at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:403)
      at org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:393)
      at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:162)
      at org.apache.spark.SparkContext.clean(SparkContext.scala:2326)
      at org.apache.spark.rdd.RDD$$anonfun$foreach$1.apply(RDD.scala:926)
      at org.apache.spark.rdd.RDD$$anonfun$foreach$1.apply(RDD.scala:925)
      at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
      at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
      at org.apache.spark.rdd.RDD.withScope(RDD.scala:363)
      at org.apache.spark.rdd.RDD.foreach(RDD.scala:925)
      ... 49 elided
    Caused by: java.io.NotSerializableException: org.apache.hadoop.hdfs.client.HdfsDataOutputStream
    Serialization stack:
        - object not serializable (class: org.apache.hadoop.hdfs.client.HdfsDataOutputStream, value: org.apache.hadoop.hdfs.client.HdfsDataOutputStream@440f55ad)
        - field (class: $iw, name: out, type: class org.apache.hadoop.fs.FSDataOutputStream)
        - object (class $iw, $iw@13c2b782)
        - field (class: $iw, name: $iw, type: class $iw)
        - object (class $iw, $iw@28aedf6e)
        - field (class: $iw, name: $iw, type: class $iw)
        - object (class $iw, $iw@14d0c3ff)
        - field (class: $iw, name: $iw, type: class $iw)
        - object (class $iw, $iw@48eb05e9)
        - field (class: $iw, name: $iw, type: class $iw)
        - object (class $iw, $iw@6b9ba1a6)
        - field (class: $iw, name: $iw, type: class $iw)
        - object (class $iw, $iw@53d519cb)
        - field (class: $iw, name: $iw, type: class $iw)
        - object (class $iw, $iw@45d7e92)
        - field (class: $iw, name: $iw, type: class $iw)
        - object (class $iw, $iw@79c1301b)
        - field (class: $line49.$read, name: $iw, type: class $iw)
        - object (class $line49.$read, $line49.$read@1a714d1)
        - field (class: $iw, name: $line49$read, type: class $line49.$read)
        - object (class $iw, $iw@79ef07b3)
        - field (class: $iw, name: $outer, type: class $iw)
        - object (class $iw, $iw@2dd246ff)
        - field (class: $anonfun$1, name: $outer, type: class $iw)
        - object (class $anonfun$1, <function1>)
      at org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:40)
      at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:46)
      at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:100)
      at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:400)
      ... 58 more

Please let me know if there is any specific way this can be achieved.

Answer 1

The foreach method on rdd requires only the arguments to be serializable. So just by writing a wrapper for ImageIO.write(x,"png", out) with serializable argument I was able to make this work.