Comment puis-je me passer d'un en-tête à partir de fichiers CSV dans Spark?

Si il y avait juste une ligne d'en-tête dans le premier enregistrement, puis le moyen le plus efficace pour filtrer serait:

rdd.mapPartitionsWithIndex {
  (idx, iter) => if (idx == 0) iter.drop(1) else iter 
}

Cela n'aide pas si bien sûr, il ya beaucoup de fichiers avec de nombreuses lignes d'en-tête à l'intérieur. Vous pouvez l'union de trois RDDs vous faites de cette manière, en effet.

Vous pourriez tout aussi bien écrire un filter qui ne correspond qu'à une ligne qui pourrait être un en-tête. C'est assez simple, mais de moins en moins efficace.

Python équivalent:

from itertools import islice

rdd.mapPartitionsWithIndex(
    lambda idx, it: islice(it, 1, None) if idx == 0 else it 
)

Spark 2.0 CSV reader est construire dans Spark, de sorte que vous pouvez facilement charger un fichier CSV comme suit:

spark.read.option("header","true").csv("filePath")

Dans PySpark vous pouvez utiliser un dataframe et de définir l'en-tête comme Vrai:

df = spark.read.csv(dataPath, header=True)

Utiliser le filter() méthode dans PySpark en filtrant le premier nom de la colonne à supprimer l'en-tête:

# Read file (change format for other file formats)
contentRDD = sc.textfile(<filepath>)

# Filter out first column of the header
filterDD = contentRDD.filter(lambda l: not l.startswith(<first column name>)

# Check your result
for i in filterDD.take(5) : print (i)

C'est une option que vous passez à la read() commande:

context = new org.apache.spark.sql.SQLContext(sc)

var data = context.read.option("header","true").csv("<path>")

//Find header from the files lying in the directory
val fileNameHeader = sc.binaryFiles("E:\\sss\\*.txt",1).map{
case (fileName, stream)=>
val header = new BufferedReader(new InputStreamReader(stream.open())).readLine()
(fileName, header)
}.collect().toMap
val fileNameHeaderBr = sc.broadcast(fileNameHeader)
//Now let's skip the header. mapPartition will ensure the header
//can only be the first line of the partition
sc.textFile("E:\\sss\\*.txt",1).mapPartitions(iter =>
if(iter.hasNext){
val firstLine = iter.next()
println(s"Comparing with firstLine $firstLine")
if(firstLine == fileNameHeaderBr.value.head._2)
new WrappedIterator(null, iter)
else
new WrappedIterator(firstLine, iter)
}
else {
iter
}
).collect().foreach(println)
class WrappedIterator(firstLine:String,iter:Iterator[String]) extends Iterator[String]{
var isFirstIteration = true
override def hasNext: Boolean = {
if (isFirstIteration && firstLine != null){
true
}
else{
iter.hasNext
}
}
override def next(): String = {
if (isFirstIteration){
println(s"For the first time $firstLine")
isFirstIteration = false
if (firstLine != null){
firstLine
}
else{
println(s"Every time $firstLine")
iter.next()
}
}
else {
iter.next()
}
}
}