Hi @Dinesh Das the following code is tested on spark-shell with scala and works perfectly with psv and csv data. the following are the datasets I used from the same directory /data/dev/spark file1.csv 1,2,3 x,y,z a,b,c file2.psv q|w|e 1|2|3 To test, you can copy paste my code into spark shell (copy only few lines/functions at a time, do not paste all code at once in Spark Shell) import org.apache.spark.{ SparkConf, SparkContext } import org.apache.spark.sql.functions.broadcast import org.apache.spark.sql.types._ import org.apache.spark.sql._ import org.apache.spark.sql.functions._ val sqlContext = new org.apache.spark.sql.SQLContext(sc) import sqlContext.implicits._ // --EDIT YOUR SCHEMA HERE case class refLineID( attr1:String, attr2:String, attr3:String ) import org.apache.hadoop.fs.{FileSystem,Path} val files = FileSystem.get( sc.hadoopConfiguration ).listStatus(new Path("/data/dev/spark")) // function to check delimiter of each file def checkDelim(file:String): String ={ val x = sc.textFile(file); val grab_x = x.take(1) // grab the first row to check delimiter val str = grab_x.mkString("") val pipe = "\\|" val comma = "\\," var delim = "" for (c <- str) { if (c == ',') { delim = comma } else if (c == '|') { delim = pipe } } return delim } // -- Function to convert RDD to dataframe after checking delimiter def convertToDF(file: String) = { var delim = "" delim = checkDelim(file) // grab the delimiter by calling function val x = sc.textFile(file); // pass the file and delimiter type to transform to dataframe val x_df = x.map(_.split(delim)) .map(a => refLineID( a(0).toString, a(1).toString, a(2).toString )).toDF x_df.show() } // -- Loop through each file and call the function 'convertToDF' files.foreach(filename => { val a = filename.getPath.toString() convertToDF(a) }) Note: I'm using Spark 1.6 and scala. I am using one function called "checkDelim" which checks the delimiter of the first row of each file under the directory. "convertToDataframe" function then knows how to split the rows and converts the data into a dataframe. Pretty simple! ... View more

Finally find out the solution. Here is the full code below. Fire up a spark shell, change the 'hadoopPath' below to your own hdfs path which contains several other directories with same schema and see it yourself. It will convert each dataset to dataframe and print the table. import org.apache.spark.{ SparkConf, SparkContext } import org.apache.spark.sql.functions.broadcast import org.apache.spark.sql.types._ import org.apache.spark.sql._ import org.apache.spark.sql.functions._ val sqlContext = new org.apache.spark.sql.SQLContext(sc) import sqlContext.implicits._ case class Test( attr1:String, attr2:String ) sc.setLogLevel("WARN") import org.apache.hadoop.fs.{FileSystem,Path} val files = FileSystem.get( sc.hadoopConfiguration ).listStatus(new Path("/hadoopPath")) def doSomething(file: String) = { println (file); // your logic of processing a single file comes here val x = sc.textFile(file); val classMapper = x.map(_.split("\\|")) .map(x => refLineID( x(0).toString, x(1).toString )).toDF classMapper.show() } files.foreach( filename => { // the following code makes sure "_SUCCESS" file name is not processed val a = filename.getPath.toString() val m = a.split("/") val name = m(10) println("\nFILENAME: " + name) if (name == "_SUCCESS") { println("Cannot Process '_SUCCSS' Filename") } else { doSomething(a) } }) ... View more