本文主要是介绍dvc 更改大型数据集的处理方法，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！
背景
2024年6月，由于需要对图像的数据集经常增删改查，故需要有一个基础和可以管理数据集的工具，查看网上相关的资料，初步尝试了dvc + git + minio的组合方式。
同时，参考官网关于dvc对于的文章大型数据集：
https://dvc.org/doc/user-guide/data-management/modifying-large-datasets
思路：
使用dvc data status --granular取得需要dvc add的图像，小步add, 减少时间。
因为每次都要自己看，故这里写了一个小脚本，自动dvc add file.
如果文件大于5000，这里直接dvc add dirs。
import os
import subprocess
import re
import time
from typing import TypedDict, Listclass ResultAddFile(TypedDict):is_need_git_commit: boolfiles_path: List[str]def get_need_add_files() -> ResultAddFile:is_need_git_commit = Falsefile_list = []# 定义要执行的命令command = "dvc data status --granular"# 正则表达式用于移除ANSI控制序列ansi_escape = re.compile(r'\x1b[^m]*m')# 使用subprocess.Popen以确保可以实时读取输出并控制编码process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)# 实时读取并打印输出is_need_add = Falseindex = 0for line in iter(process.stdout.readline, b''):# 显式指定使用UTF-8解码index += 1# 去掉尾的空格text = line.decode('utf-8', 'ignore').strip()# 移除ANSI控制序列text = ansi_escape.sub('', text).strip()match = re.search(r':\s*(.*)', text)if match:file_path = match.group(1).strip()if "." not in file_path:continueif is_need_add:print(f"{index}：{text}")file_list.append(file_path)else:if len(text) == 0:continueprint(f"{index}：{text}")if 'dvc commit' in text:is_need_add = Trueif 'git status' in text:is_need_git_commit = True# 等待命令执行完毕并获取返回码process.wait()return_code = process.returncodeif return_code == 0:print(f"文件数量：{len(file_list)}")else:print(f"失败，返回码：{return_code}")# 创建并返回 ResultAddFile 类型的结果result = {"is_need_git_commit": is_need_git_commit,"files_path": file_list}return resultdef dvc_add(_file_list):for _file_path in _file_list:# 确保文件路径适合在命令行中使用，这一步可能不是必须的，取决于具体的系统和命令行工具encoded_path = os.fsencode(_file_path) # 如果dvc需要字节串command = f"dvc add {encoded_path.decode('utf-8')}" # 或者直接使用file_path，根据dvc的实际要求print(command)# 记录命令开始执行的时间start_time = time.time()# 使用Popen执行命令# 使用subprocess.Popen以确保可以实时读取输出并控制编码process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True,encoding='utf-8')stdout, stderr = process.communicate()return_code = process.returncode# os.system(command)# 计算命令执行的耗时（转换为毫秒）execution_time = (time.time() - start_time)print(f"命令耗时: {execution_time:.2f}秒")if return_code == 0:print(f"命令成功执行耗时: {execution_time:.2f}秒")else:print(f"命令执行失败，返回码：{return_code}\n错误信息：{stderr} 执行耗时: {execution_time:.2f}毫秒")if __name__ == '__main__':# 假设file_list已经是一个包含超过10000个文件路径的列表unique_paths = set() # 用于存储唯一的文件路径result: ResultAddFile = get_need_add_files()file_list = result["files_path"]is_need_git_commit = result["is_need_git_commit"]if len(file_list) > 5000:for file_path in file_list:dir_path = os.path.dirname(file_path)if dir_path not in unique_paths:unique_paths.add(dir_path)for dir_path in unique_paths:print(dir_path)# 确保文件路径适合在命令行中使用，这一步可能不是dvc_add(unique_paths)elif len(file_list) > 0:dvc_add(file_list)# 最后的提示if is_need_git_commit:print("有文件未提交，请执行: git commit!")这篇关于dvc 更改大型数据集的处理方法的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！
