如何使用PySpark计算ADLS中的目录大小?

dm7nw8vv  于 2022-12-13  发布在  Apache
关注(0)|答案(5)|浏览(146)

我想计算一个包含子文件夹和子文件的目录(例如- XYZ)大小。我想要所有文件和XYZ中的所有内容的总大小。
我可以找出一个特定路径内的所有文件夹。但我想所有的大小在一起。我还看到
display(dbutils.fs.ls("/mnt/datalake/.../XYZ/.../abc.parquet"))
给我abc文件的数据大小。但我想要完整的XYZ大小。

vc6uscn9

vc6uscn91#

dbutils.fs.ls不像cpmvrm那样具有递归功能。因此,您需要自己迭代。下面是一个代码片段,它将为您完成此任务。从数据块笔记本运行代码。

from dbutils import FileInfo
from typing import List

root_path = "/mnt/datalake/.../XYZ"

def discover_size(path: str, verbose: bool = True):
  def loop_path(paths: List[FileInfo], accum_size: float):
    if not paths:
      return accum_size
    else:
      head, tail = paths[0], paths[1:]
      if head.size > 0:
        if verbose:
          print(f"{head.path}: {head.size / 1e6} MB")
        accum_size += head.size / 1e6
        return loop_path(tail, accum_size)
      else:
        extended_tail = dbutils.fs.ls(head.path) + tail
        return loop_path(extended_tail, accum_size)

  return loop_path(dbutils.fs.ls(path), 0.0)

discover_size(root_path, verbose=True)  # Total size in megabytes at the end

如果该位置是在dbfs中挂载的。那么您可以使用du -h方法(尚未测试)。如果您在笔记本中,请使用以下内容创建一个新单元:

%sh
du -h /mnt/datalake/.../XYZ
epggiuax

epggiuax2#

@Emer的答案很好,但是可以很快找到RecursionError: maximum recursion depth exceeded,因为它对每个文件都进行递归(如果有X个文件,则会有X个叠瓦递归)。
以下是仅针对文件夹的递归:

%python
from dbutils import FileInfo
from typing import List

def discover_size2(path: str, verbose: bool = True):
  def loop_path(path: str):
    accum_size = 0.0
    path_list = dbutils.fs.ls(path)
    if path_list:
      for path_object in path_list:
        if path_object.size > 0:
          if verbose:
            print(f"{path_object.path}: {path_object.size / 1e6} MB")
          accum_size += path_object.size / 1e6
        else:
          # Folder: recursive discovery
          accum_size += loop_path(path_object.path)
    return accum_size

  return loop_path(path)
pobjuy32

pobjuy323#

尝试使用dbutils ls命令,获取 Dataframe 中的文件列表,然后对size列使用聚合函数SUM()进行查询:

val fsds = dbutils.fs.ls("/mnt/datalake/.../XYZ/.../abc.parquet").toDF

fsds.createOrReplaceTempView("filesList")

display(spark.sql("select COUNT(name) as NoOfRows, SUM(size) as sizeInBytes from fileListPROD"))
qlckcl4x

qlckcl4x4#

爱死answer by Emer了!
少量添加:
如果你遇见
“模块未找到错误:没有名为'dbutils'的模块“
试试这个from dbruntime.dbutils而不是from dbutils,它对我很有效!

ldioqlga

ldioqlga5#

对于那些仍在用@robin loce的方法达到递归极限的人,这里有一个纯粹的迭代答案:

# from dbutils import FileInfo # Not required in databricks
# from dbruntime.dbutils import FileInfo # may work for some people

def get_size_of_path(path):
    return sum([file.size for file in get_all_files_in_path(path)])

def get_all_files_in_path(path, verbose=False):
    nodes_new = []

    nodes_new = dbutils.fs.ls(path)
    files = []

    while len(nodes_new) > 0:
        current_nodes = nodes_new
        nodes_new = []
        for node in current_nodes:
            if verbose:
                print(f"Processing {node.path}")
            children = dbutils.fs.ls(node.path)
            for child in children:
                if child.size == 0 and child.path != node.path:
                    nodes_new.append(child)
                elif child.path != node.path:
                    files.append(child)
    return files

path = "s3://some/path/"

print(f"Size of {path} in gb: {get_size_of_path(path) / 1024 / 1024 / 1024}")

相关问题