Hi @Maher Hattabi
For Spark 1.6+
What you need to do is load all the csv files with a for loop in a batch processing manner. As u inject the same schema into each of them, convert them to a dataframe; union each of the dataframe in another var. In that way, all of them will be just one dataframe. Following code does the work. You can follow my code and test them in spark-shell
Contents of file1.csv
x,y,z
Contents of file2.csv
a,b,c
c,d,e
Store them in a hdfs directory and change the path according to yours in the following code where it says 'hadoopPath'
NOTE: While working on spark-shell, don't paste all code at once. It yields error sometimes. Paste once bunch at a time.
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 Test(
attr1:String,
attr2:String,
attr3:String
)
import org.apache.hadoop.fs.{FileSystem,Path}
/*
initialize empty dataframe
(This dataframe will be the final one where we union all others)
*/
var all_df = Seq.empty[Test].toDF
// -- EDIT YOUR HDFS PATH HERE 'hadoopPath'
val files = FileSystem.get( sc.hadoopConfiguration ).listStatus(new Path("/hadoopPath/"))
// -- Function for all operations to be executed in each file iteration
def convertToDFandUnion(file: String) = {
val x = sc.textFile(file);
val x_df = x.map(_.split("\\,"))
.map(a => refLineID(
a(0).toString,
a(1).toString,
a(2).toString
)).toDF
x_df.show()
// This is where we make each dataframe into one
all_df = all_df.unionAll(x_df)
all_df.show()
}
// -- Loop through each file and call the function 'ConvertToDFAnd Union'
files.foreach(filename => {
val a = filename.getPath.toString()
convertToDFandUnion(a)
})
