### HDFS目录下所有ZIP文件及其大小的Python脚本实现 为了通过 Python 和 `subprocess` 模块递归获取 HDFS 路径 `/data/test/` 下的所有 ZIP 文件列表及每个文件的容量大小，可以按照以下方式编写脚本： #### 实现思路 1. 使用 `hdfs dfs -ls` 命令递归列出指定路径下的所有文件。 2. 过滤出扩展名为 `.zip` 的文件。 3. 获取这些文件的大小。 以下是完整的 Python 脚本实现： ```python import subprocess def get_hdfs_zip_files(hdfs_path): """ 递归获取HDFS路径下所有的.zip文件及其大小 参数: hdfs_path (str): 需要查询的HDFS路径 返回: list: 包含字典的列表，每个字典表示一个.zip文件的信息{'name': '文件名', 'size': 大小} """ try: # 执行命令以递归列出HDFS上的所有文件 result = subprocess.run( ["hdfs", "dfs", "-ls", "-R", hdfs_path], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True ) if result.returncode != 0: raise Exception(f"Error executing command: {result.stderr}") lines = result.stdout.splitlines() zip_files = [] for line in lines: parts = line.split() # 将每行按空格分割成多个部分 if len(parts) >= 8 and parts[-1].endswith(".zip"): # 判断是否为.zip文件 file_size = int(parts[4]) # 提取文件大小 file_name = parts[-1] # 提取文件全路径 zip_files.append({ "name": file_name, # 文件名称 "size": file_size # 文件大小（单位：字节） }) return zip_files except Exception as e: print(f"An error occurred while fetching .zip files from HDFS: {e}") return [] # 测试函数 if __name__ == "__main__": path = "/data/test/" # 替换为目标HDFS路径 zip_files_info = get_hdfs_zip_files(path) if zip_files_info: print("Found the following .zip files:") for info in zip_files_info: print(f"{info['name']} -> Size: {info['size']} bytes") else: print("No .zip files found or an error occurred.") ``` --- #### 关键点说明 1. **`subprocess.run` 方法** 使用该方法调用外部 Shell 命令来交互 HDFS。这里使用的是 `hdfs dfs -ls -R /path/to/dir` 来递归列出目标路径中的所有文件[^1]。 2. **过滤逻辑** 对于每一行输出，判断最后一列（即文件路径）是否以 `.zip` 结尾，并提取对应的文件大小字段。 3. **异常处理** 如果命令执行失败或者返回码不为零，则抛出异常并记录错误信息。 4. **结果存储** 将符合条件的文件信息存入列表中，最终返回给调用者。 --- #### 输出示例 假设 HDFS 中存在如下结构： ``` /data/test/ ├── folder1/file1.zip (1024 字节) ├── folder1/subfolder/file2.zip (2048 字节) └── other.txt (512 字节) ``` 运行脚本后可能得到的结果为： ``` Found the following .zip files: /data/test/folder1/file1.zip -> Size: 1024 bytes /data/test/folder1/subfolder/file2.zip -> Size: 2048 bytes ``` ---