将csv转换为RDD答案

【问题标题】：Convert csv to RDD将csv转换为RDD
【发布时间】：2015-02-15 08:52:57
【问题描述】：

我在How do I convert csv file to rdd尝试了接受的解决方案，我想打印除“om”之外的所有用户：

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") != "om") // filter the header out
val users = rows.map(row => header(row,"user")
users.collect().map(user => println(user))

但我得到了一个错误：

java.util.NoSuchElementException: key not found: user

我尝试调试它并在header 中找到index 属性，如下所示：

由于我是 spark 和 scala 的新手，这是否意味着 user 已经在 Map 中？那为什么会出现key not found 错误呢？

【问题讨论】：

我无法重现您的问题。在 REPL 中，它返回：Array[String] = Array(user, daniel, 3754978)。顺便说一句，我猜你仍然想通过 data.filter(line => header(line,"user") != "user")) 过滤掉标题
你能在文件中添加第一行数据吗？
@maasg，感谢您的 cmets。我发现了我的错误，这不是 spark/scala 的问题，这是我创建 csv 时犯的错误，我将在下面的答案中解释更多。

标签： scala csv apache-spark

【解决方案1】：

我发现了我的错误。它与 Spark/Scala 无关。当我创建示例 csv 时，我在 R 中使用命令：

df <- data.frame(user=c('om','daniel','3754978'),topic=c('scala','spark','spark'),hits=c(120,80,1))
write.csv(df, "df.csv",row.names=FALSE)

但是write.csv默认会在因子周围添加"，所以这就是为什么地图找不到关键user因为"user"是真正的关键，使用

write.csv(df, "df.csv",quote=FALSE, row.names=FALSE)

会解决这个问题。

【讨论】：

【解决方案2】：

我已重写示例代码以删除标头方法。 IMO，此示例提供了更易于遵循的逐步演练。 Here is a more detailed explanation.

def main(args: Array[String]): Unit = {
  val csv = sc.textFile("/path/to/your/file.csv")

  // split / clean data
  val headerAndRows = csv.map(line => line.split(",").map(_.trim))
  // get header
  val header = headerAndRows.first
  // filter out header
  val data = headerAndRows.filter(_(0) != header(0))
  // splits to map (header/value pairs)
  val maps = data.map(splits => header.zip(splits).toMap)
  // filter out the 'om' user
  val result = maps.filter(map => map("user") != "om")
  // print result
  result.foreach(println)
}

【讨论】：