如何使用 FileSystem API 计算分区？

Posted 2023-04-15

【中文标题】如何使用 FileSystem API 计算分区？

【英文标题】：How to count partitions with FileSystem API?

【发布时间】：2019-10-22 22:25:12

【问题描述】：

我正在使用 Hadoop 2.7 版及其文件系统 API。问题是关于“如何使用 API 计算分区？” 但是，把它放到软件问题中，我在这里处理一个 Spark-Shell 脚本......关于脚本的具体问题是

下面的变量parts是在统计表分区的数量，还是别的什么？

import org.apache.hadoop.fs.FileSystem, Path
import scala.collection.mutable.ArrayBuffer
import spark.implicits._

val warehouse = "/apps/hive/warehouse"  // the Hive default location for all databases
val db_regex  = """\.db$""".r   // filter for names like "*.db"
val tab_regex = """\.hive\-staging_""".r    // signature of Hive files

val trStrange = "[\\s/]+|[^\\x00-\\x7F]+|[\\pCntrl&&[^\r\n\t]]+|\\pC+".r //mark
def cutPath (thePath: String, toCut: Boolean = true) : String =
  if (toCut) trStrange.replaceAllIn( thePath.replaceAll("^.+/", ""),  "@") else thePath

val fs_get = FileSystem.get( sc.hadoopConfiguration )
fs_get.listStatus( new Path(warehouse) ).foreach( lsb => 
    val b = lsb.getPath.toString
    if (db_regex.findFirstIn(b).isDefined) 
       fs_get.listStatus( new Path(b) ).foreach( lst => 
            val lstPath = lst.getPath
            val t = lstPath.toString
            var parts = -1
            var size = -1L
            if (!tab_regex.findFirstIn(t).isDefined) 
              try 
                  val pp = fs_get.listStatus( lstPath )
                  parts = pp.length // !HERE! partitions?
                  pp.foreach( p => 
                     try  // SUPPOSING that size is the number of bytes of table t
                        size  = size  + fs.getContentSummary(p.getPath).getLength
                      catch  case _: Throwable => 
                  )
               catch  case _: Throwable =>  
              println(s"$cutPath(b),$cutPath(t),$parts,$size")
            
        )
) // x warehouse loop
System.exit(0)  // get out from spark-shell

这只是一个展示重点的例子：Hive 默认数据库 FileSystem 结构的正确扫描和语义解释，使用 Hive FileSystem API。该脚本有时需要一些内存，但工作正常。使用 sshell --driver-memory 12G --executor-memory 18G -i teste_v2.scala > output.csv

运行

---

注意：这里的目的不是通过任何其他方法（例如 HQL DESCRIBE 或 Spark Schema）对分区进行计数，而是使用 API 来实现...用于控制和数据质量 检查，API 作为一种“低级测量”很重要。

【参考方案1】：

Hive 将其元数据构建为 数据库 > 表 > 分区 > 文件。这通常转化为文件系统目录结构<hive.warehouse.dir>/database.db/table/partition/.../files。其中/partition/.../ 表示表可以由多个列分区，从而创建嵌套级别的子目录。 （partition 是一个 directory，按照惯例名称为 .../partition_column=value）。

因此，如果我没记错的话，您的脚本似乎将打印每个数据库中每个单列分区表的文件数 (parts) 及其总长度 (size)。

作为替代方案，我建议您查看hdfs dfs -count 命令，看看它是否适合您的需求，并可能将其包装在一个简单的 shell 脚本中以循环访问数据库和表。