DVC：数据版本控制工具

import hashlibimport osimport json def compute_file_hash(filepath):    hasher = hashlib.md5()    with open(filepath, 'rb') as f:        while chunk := f.read(8192):            hasher.update(chunk)    return hasher.hexdigest() def dvc_add(filepath):    md5 = compute_file_hash(filepath)    cache_dir = '.dvc/cache'    cache_subdir = md5[:2]    cache_path = os.path.join(cache_dir, cache_subdir, md5)    os.makedirs(os.path.dirname(cache_path), exist_ok=True)    if not os.path.exists(cache_path):        os.rename(filepath, cache_path)    dvc_file = filepath + '.dvc'    meta = {'outs': [{'path': os.path.basename(filepath), 'md5': md5, 'size': os.path.getsize(cache_path)}]}    with open(dvc_file, 'w') as f:        json.dump(meta, f, indent=2)    with open('.gitignore', 'a') as f:        f.write(filepath + '\n')    return md5 hash_val = dvc_add('dataset.csv')print(f'DVC add完成, MD5: {hash_val}')

import subprocessimport json def run_cmd(cmd):    result = subprocess.run(cmd, shell=True, capture_output=True, text=True)    return result.stdout.strip(), result.stderr.strip() def dvc_workflow_demo():    # 1. dvc add - 将数据纳入DVC管理    out, err = run_cmd('dvc add data/dataset.csv')    print(f'dvc add: {out}')        # 2. git add .dvc文件和.gitignore    out, err = run_cmd('git add data/dataset.csv.dvc .gitignore')    out, err = run_cmd('git commit -m "Update dataset to v2"')    print(f'git commit: {out}')        # 3. dvc push - 推送数据到远程存储    out, err = run_cmd('dvc push')    print(f'dvc push: {out}')        # 4. dvc checkout - 恢复特定版本数据    out, err = run_cmd('git checkout HEAD~1 -- data/dataset.csv.dvc')    out, err = run_cmd('dvc checkout')    print(f'dvc checkout: {out}')        return 'DVC workflow完成' print(dvc_workflow_demo())

import yamlimport subprocess def create_dvc_yaml():    pipeline = {        'stages': {            'prepare_data': {                'cmd': 'python src/prepare.py --input data/raw.csv --output data/prepared.csv',                'deps': ['data/raw.csv', 'src/prepare.py'],                'outs': ['data/prepared.csv']            },            'train_model': {                'cmd': 'python src/train.py --data data/prepared.csv --model models/model.pkl',                'deps': ['data/prepared.csv', 'src/train.py'],                'outs': ['models/model.pkl'],                'metrics': [{'metrics.json': {'cache': False}}]            },            'evaluate': {                'cmd': 'python src/evaluate.py --model models/model.pkl --output eval.json',                'deps': ['models/model.pkl', 'src/evaluate.py'],                'outs': ['eval.json']            }        }    }    with open('dvc.yaml', 'w') as f:        yaml.dump(pipeline, f, default_flow_style=False)    return 'dvc.yaml created' def run_pipeline():    result = subprocess.run(['dvc', 'repro'], capture_output=True, text=True)    print(result.stdout)    return 'Pipeline executed' print(create_dvc_yaml())print(run_pipeline())

import hashlibimport osimport jsonimport shutilimport yaml CACHE_DIR = '.dvc/cache'REMOTE_DIR = '/tmp/dvc_remote' def compute_md5(filepath):    hasher = hashlib.md5()    with open(filepath, 'rb') as f:        while chunk := f.read(8192):            hasher.update(chunk)    return hasher.hexdigest() def dvc_add(filepath):    md5 = compute_md5(filepath)    cache_path = os.path.join(CACHE_DIR, md5[:2], md5)    os.makedirs(os.path.dirname(cache_path), exist_ok=True)    if not os.path.exists(cache_path):        shutil.copy2(filepath, cache_path)    dvc_meta = {        'outs': [{            'path': os.path.basename(filepath),            'md5': md5,            'size': os.path.getsize(filepath)        }]    }    dvc_file = filepath + '.dvc'    with open(dvc_file, 'w') as f:        json.dump(dvc_meta, f, indent=2)    with open('.gitignore', 'a') as f:        if filepath not in open('.gitignore').read():            f.write(filepath + '\n')    return md5 def dvc_push(md5):    cache_path = os.path.join(CACHE_DIR, md5[:2], md5)    remote_path = os.path.join(REMOTE_DIR, md5[:2], md5)    os.makedirs(os.path.dirname(remote_path), exist_ok=True)    if not os.path.exists(remote_path):        shutil.copy2(cache_path, remote_path)        print(f'Pushed {md5[:8]}... to remote')    else:        print(f'Skip {md5[:8]}... (already exists)') def dvc_pull(dvc_file):    with open(dvc_file) as f:        meta = json.load(f)    md5 = meta['outs'][0]['md5']    original_path = meta['outs'][0]['path']    remote_path = os.path.join(REMOTE_DIR, md5[:2], md5)    cache_path = os.path.join(CACHE_DIR, md5[:2], md5)    os.makedirs(os.path.dirname(cache_path), exist_ok=True)    if not os.path.exists(cache_path):        shutil.copy2(remote_path, cache_path)    shutil.copy2(cache_path, original_path)    print(f'Pulled {original_path} (md5: {md5[:8]}...)')    return original_path def create_pipeline_yaml():    pipeline = {        'stages': {            'prepare': {                'cmd': 'python prepare.py',                'deps': ['data/raw.csv', 'prepare.py'],                'outs': ['data/prepared.csv']            },            'train': {                'cmd': 'python train.py',                'deps': ['data/prepared.csv', 'train.py'],                'outs': ['model.pkl'],                'metrics': [{'metrics.json': {'cache': False}}]            }        }    }    with open('dvc.yaml', 'w') as f:        yaml.dump(pipeline, f, default_flow_style=False)    return pipeline def main():    os.makedirs('data', exist_ok=True)    with open('data/raw.csv', 'w') as f:        f.write('feature1,feature2,label\n1.0,2.0,0\n3.0,4.0,1\n')    md5_v1 = dvc_add('data/raw.csv')    print(f'Version 1 MD5: {md5_v1}')    dvc_push(md5_v1)    with open('data/raw.csv', 'w') as f:        f.write('feature1,feature2,label\n1.0,2.0,0\n3.0,4.0,1\n5.0,6.0,1\n')    md5_v2 = dvc_add('data/raw.csv')    print(f'Version 2 MD5: {md5_v2}')    dvc_push(md5_v2)    print(f'\nData changed: {md5_v1 != md5_v2}')    pipeline = create_pipeline_yaml()    print(f'\nPipeline stages: {list(pipeline["stages"].keys())}')    dvc_pull('data/raw.csv.dvc')    print('\nDVC workflow demo complete') if __name__ == '__main__':    main()