如何在sparkscalashell中列出hdfs位置中的所有csv文件？

9wbgstp7 于 2021-05-29 发布在 Hadoop

关注(0)|答案(3)|浏览(305)

这样做的目的是为了在hdfs中的第二个位置操作和保存每个数据文件的副本。我将使用

RddName.coalesce(1).saveAsTextFile(pathName)

将结果保存到hdfs。
这就是为什么我想做每一个文件分开，即使我相信性能将不那么有效。但是，我还没有决定如何将csv文件路径列表存储到一个字符串数组中，然后使用单独的rdd循环每个字符串。
让我们使用以下匿名示例作为hdfs源位置：

/data/email/click/date=2015-01-01/sent_20150101.csv
/data/email/click/date=2015-01-02/sent_20150102.csv
/data/email/click/date=2015-01-03/sent_20150103.csv

我知道如何使用hadoop fs shell列出文件路径：

HDFS DFS -ls /data/email/click/*/*.csv

我知道如何为所有数据创建一个rdd：

val sentRdd = sc.textFile( "/data/email/click/*/*.csv" )

hadoop hdfs scala apache-spark

来源：https://stackoverflow.com/questions/32771089/how-can-one-list-all-csv-files-in-an-hdfs-location-within-the-spark-scala-shell

3条答案

按热度按时间

pvabu6sv1#

这就是最终对我起作用的原因：

import org.apache.hadoop.fs._
import org.apache.spark.deploy.SparkHadoopUtil
import java.net.URI

val hdfs_conf = SparkHadoopUtil.get.newConfiguration(sc.getConf)
val hdfs = FileSystem.get(hdfs_conf)
// source data in HDFS
val sourcePath = new Path("/<source_location>/<filename_pattern>")

hdfs.globStatus( sourcePath ).foreach{ fileStatus =>
   val filePathName = fileStatus.getPath().toString()
   val fileName = fileStatus.getPath().getName()

   // < DO STUFF HERE>

} // end foreach loop

赞(0）回复(0）举报 2021-05-30

pexxcrt22#

我还没有完全测试过，但类似的方法似乎很有效：

import org.apache.spark.deploy.SparkHadoopUtil
import org.apache.hadoop.fs.{FileSystem, Path, LocatedFileStatus, RemoteIterator}
import java.net.URI

val path: String = ???

val hconf = SparkHadoopUtil.get.newConfiguration(sc.getConf)
val hdfs = FileSystem.get(hconf)
val iter = hdfs.listFiles(new Path(path), false)

def listFiles(iter: RemoteIterator[LocatedFileStatus]) = {
  def go(iter: RemoteIterator[LocatedFileStatus], acc: List[URI]): List[URI] = {
    if (iter.hasNext) {
      val uri = iter.next.getPath.toUri
      go(iter, uri :: acc)
    } else {
      acc
    }
  }
  go(iter, List.empty[java.net.URI])
}

listFiles(iter).filter(_.toString.endsWith(".csv"))

赞(0）回复(0）举报 2021-05-30

aij0ehis3#

sc.wholetextfiles（路径）应该有帮助。它给出（filepath，filecontent）的rdd。

赞(0）回复(0）举报 2021-05-30