How do I convert csv file to rdd

by

A simplistic approach would be to have a way to preserve the header.

Let’s say you have a file.csv like:

user, topic, hits
om,  scala, 120
daniel, spark, 80
3754978, spark, 1

We can define a header class that uses a parsed version of the first row:

class SimpleCSVHeader(header:Array[String]) extends Serializable {
  val index = header.zipWithIndex.toMap
  def apply(array:Array[String], key:String):String = array(index(key))
}

That we can use that header to address the data further down the road:

val csv = sc.textFile("file.csv")  // original file
val data = csv.map(line => line.split(",").map(elem => elem.trim)) //lines in rows
val header = new SimpleCSVHeader(data.take(1)(0)) // we build our header with the first line
val rows = data.filter(line => header(line,"user") != "user") // filter the header out
val users = rows.map(row => header(row,"user")
val usersByHits = rows.map(row => header(row,"user") -> header(row,"hits").toInt)
...

Note that the header is not much more than a simple map of a mnemonic to the array index. Pretty much all this could be done on the ordinal place of the element in the array, like user = row(0)

PS: Welcome to Scala 🙂

More Related Contents:

  1. How to define and use a User-Defined Aggregate Function in Spark SQL?
  2. (Why) do we need to call cache or persist on a RDD
  3. How to load local file in sc.textFile, instead of HDFS
  4. Automatically and Elegantly flatten DataFrame in Spark SQL
  5. How do I skip a header from CSV files in Spark?
  6. Spark extracting values from a Row
  7. Stratified sampling in Spark
  8. Caused by: java.lang.NullPointerException at org.apache.spark.sql.Dataset
  9. Split 1 column into 3 columns in spark scala
  10. Provide schema while reading csv file as a dataframe
  11. How to flatmap a nested Dataframe in Spark
  12. How to save DataFrame directly to Hive?
  13. About how to add a new column to an existing DataFrame with random values in Scala
  14. How to use Column.isin with list?
  15. Function returns an empty List in Spark
  16. How to compare two dataframe and print columns that are different in scala
  17. Operate on neighbor elements in RDD in Spark
  18. Customize SparkContext using sparkConf.set(..) when using spark-shell
  19. Renaming column names of a DataFrame in Spark Scala
  20. What is going wrong with `unionAll` of Spark `DataFrame`?
  21. Spark: long delay between jobs
  22. How do I iterate RDD’s in apache spark (scala)
  23. How to get keys and values from MapType column in SparkSQL DataFrame
  24. How to compute cumulative sum using Spark
  25. Addition of two RDD[mllib.linalg.Vector]’s
  26. Add Number of days column to Date Column in same dataframe for Spark Scala App
  27. How to prevent java.lang.OutOfMemoryError: PermGen space at Scala compilation?
  28. How to create DataFrame from Scala’s List of Iterables?
  29. How to create a Dataset of Maps?
  30. Spark ML VectorAssembler returns strange output

Leave a Comment