scala - 从Scala中的HDFS加载.csv文件

Question

所以我基本上有以下代码来读取.csv文件并将其存储在Array[Array[String]]中:

def load(filepath: String): Array[Array[String]] = {
      var data = Array[Array[String]]()
      val bufferedSource = io.Source.fromFile(filepath)
      for (line <- bufferedSource.getLines) {
        data :+ line.split(",").map(_.trim)
      }
      bufferedSource.close
      return data.slice(1,data.length-1) //skip header
  }

适用于未存储在HDFS上的文件。但是，当我在HDFS上尝试相同的操作时，我得到

No such file or directory found

在HDFS上写入文件时，我还必须更改原始代码，并向 FileSystem添加一些 Path和 PrintWriter参数，但是这次我完全不知道该怎么做。

我到此为止:

  def load(filepath: String, sc: SparkContext): Array[Array[String]] = {
      var data = Array[Array[String]]()
      val fs = FileSystem.get(sc.hadoopConfiguration)
      val stream = fs.open(new Path(filepath))
      var line = ""
      while ((line = stream.readLine()) != null) {
        data :+ line.split(",").map(_.trim)
      }

      return data.slice(1,data.length-1) //skip header
  }

这应该可以，但是在将行与null进行比较或者其长度超过0时，我得到了 NullPointerException。

Answer 1

这段代码将从HDFS中读取一个.csv文件:

  def read(filepath: String, sc: SparkContext): ArrayBuffer[Array[String]] = {
      var data = ArrayBuffer[Array[String]]()
      val fs = FileSystem.get(sc.hadoopConfiguration)
      val stream = fs.open(new Path(filepath))
      var line = stream.readLine()
      while (line != null) {
        val row = line.split(",").map(_.trim)
        data += row
        line = stream.readLine()
      }
      stream.close()

      return data // or return data.slice(1,data.length-1) to skip header
  }