加载a。Scala中HDFS的csv文件

所以我基本上有以下代码来阅读。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上尝试同样的方法时,我得到

找不到此类文件或目录

在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
  }

这应该行得通,但我有一个 NullPointerException 将行与null进行比较时,或者如果其长度超过0。