1

I am writing a spark/scala program to read in ZIP files, unzip them and write the contents to a set of new files. I can get this to work for writing to the local file system but wondered if there was a way to to write the output files to a distributed file system such as HDFS. Code is shown below`

import java.util.zip.ZipInputStream
import org.apache.spark.input.PortableDataStream
import java.io._

var i =1
sc.binaryFiles("file:///d/tmp/zips/").flatMap((file:(String, PortableDataStream)) => {   
   val zipStream = new ZipInputStream(file._2.open)            
   val entry = zipStream.getNextEntry                            
   val iter = scala.io.Source.fromInputStream(zipStream).getLines          
   val fname = f"/d/tmp/myfile$i.txt" 

   i = i + 1

   val xx = iter.mkString
   val writer = new PrintWriter(new File(fname))
   writer.write(xx)
   writer.close()

   iter                                                       
}).collect()

`

4 Answers 4

5

You can easy write data to HDFS using hadoop-common library (if you are using sbt as dependency manangement tool, add thath library to your dependency). With that you can create a FileSystem object :

 private val fs = {
    val conf = new Configuration()
    FileSystem.get(conf)
  }

Be sure to configure the FileSystem with your hadoop cluster information (core-site.xml, etc)

Then you can write, for example a String to path (in your case you should deal with streams), on HDFS as following:

@throws[IOException]
  def writeAsString(hdfsPath: String, content: String) {
    val path: Path = new Path(hdfsPath)
    if (fs.exists(path)) {
      fs.delete(path, true)
    }
    val dataOutputStream: FSDataOutputStream = fs.create(path)
    val bw: BufferedWriter = new BufferedWriter(new OutputStreamWriter(dataOutputStream, "UTF-8"))
    bw.write(content)
    bw.close
  }
2
  • to what does this FileSystem belongs to ? java.nio doesn't seem be it.
    – ss301
    Commented Dec 31, 2021 at 7:37
  • 1
    It's the Hadoop FileSystem, so you need to bring it to your dependency
    – dumitru
    Commented Jan 12, 2022 at 8:55
1
sc.binaryFiles("/user/example/zip_dir", 10)                   //make an RDD from *.zip files in HDFS
    .flatMap((file: (String, PortableDataStream)) => {        //flatmap to unzip each file
        val zipStream = new ZipInputStream(file._2.open)      //open a java.util.zip.ZipInputStream
        val entry = zipStream.getNextEntry                    //get the first entry in the stream
        val iter = Source.fromInputStream(zipStream).getLines //place entry lines into an iterator
        iter.next                                             //pop off the iterator's first line
        iter                                                  //return the iterator
    })
    .saveAsTextFile("/user/example/quoteTable_csv/result.csv")
0

You should have a look at the method saveAsTextFile from the official documentation : http://spark.apache.org/docs/latest/programming-guide.html

It will allow you to save to HDFS :

iter.saveAsTextFile("hdfs://...")
3
  • In that code iter is not a RDD, so cannot write it. Maybe with a conversion first.
    – dumitru
    Commented Feb 17, 2017 at 10:59
  • Yes I think a cast would we be good here. RDD should be the data type to manipulate over spark in order to get distributed data over the cluster.
    – chateaur
    Commented Feb 17, 2017 at 11:03
  • That's the nub of the issue. I have tried everything I can think of to get the data in my iter to an RDD to enable the use of saveasTextFile but have came up short. If anyone has solved this please let me know Commented Feb 17, 2017 at 11:24
0

You can try saveAsTextFile method.

Write the elements of the dataset as a text file (or set of text files) in a given directory in the local filesystem, HDFS or any other Hadoop-supported file system. Spark will call toString on each element to convert it to a line of text in the file.

It will save each partition as a different file, The number of partition you will end up with will be the same as the number of your input files, unless you repartition or coalesce.

3
  • Please see my above comments as to why using saveasTextFile is an issue Commented Feb 17, 2017 at 11:25
  • Can't You can write the whole RDD not each file separately. instead of collect use saveAsText file ? Commented Feb 17, 2017 at 11:31
  • That concatenates all the unzipped data for each together into one file. That's not what I want. I want each unzipped file to be in its own separate file Commented Feb 17, 2017 at 11:36

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.