我正在使用斯卡拉。
我想過濾最新的檔案夾并從 hdfs 目錄中讀取最新的以及其中的所有檔案。
現在看起來像
val read_csv =
spark
.read
.format("csv")
.load( "hdfs://device/signs/load=16?)
在檔案夾標志中，很少有帶有負載的檔案夾（負載 = 10，負載 = 13，負載 = 14，負載 = 16），我只想獲得最大值。
uj5u.com熱心網友回復：
試試這個方法：
private def getMaxPartitionValue(path: String, partitionName: String, sparkSession: SparkSession): String = {
// This will create a Path object using the hadoop configuration used in the spark session
val hdfs = new Path(path).getFileSystem(sparkSession.sparkContext.hadoopConfiguration)
// This will filter all folders that starts with the partition name, in your case load=
val partitionNames = hdfs.listStatus(new Path(path)).toList.filter(_.getPath.getName.startsWith(partitionName "="))
// if no partition starts with load= the we return an empty string
if (partitionNames.isEmpty) return ""
// this is to only get the partition value ( everything after load=) and then get the max value of all these partitions
partitionNames.map(_.getPath.getName).sorted.reverse.head
}
像這樣稱呼它：
val maxPartitionPath = path File.separator getMaxPartitionValue(path, "load", sparkSession)
uj5u.com熱心網友回復：
如果您沒有在堆疊中使用 hive 元存盤，您可以使用 Hadoop FS API 找到最新的磁區。您可以訪問檔案元資料并以您需要的方式處理它：
import org.apache.hadoop.fs.{FileSystem, Path}
val fs = FileSystem.get(sparkSession.sparkContext.hadoopConfiguration)
val ls = fs.listStatus(new Path("/"))
如果您的集群上有 Metastore，則可以使用 sql 訪問磁區
sparkSession.sql("SHOW PARTITIONS tablename")
轉載請註明出處，本文鏈接：https://www.uj5u.com/gongcheng/519542.html