fix: 修复存储和加载逻辑
This commit is contained in:
+24
-26
@@ -1,6 +1,6 @@
|
|||||||
from typing import Dict, List, Callable, Tuple, Union
|
from pathlib import Path
|
||||||
|
from typing import Dict, List, Callable, Union
|
||||||
from datasets import DatasetDict
|
from datasets import DatasetDict
|
||||||
import numpy as np
|
|
||||||
from modules.tokenizer import BpeTokenizer
|
from modules.tokenizer import BpeTokenizer
|
||||||
from tqdm import tqdm
|
from tqdm import tqdm
|
||||||
from torch import Tensor
|
from torch import Tensor
|
||||||
@@ -16,7 +16,6 @@ def fetch_files(directory):
|
|||||||
return [os.path.join(root, f)
|
return [os.path.join(root, f)
|
||||||
for root, _, files in os.walk(directory) for f in files]
|
for root, _, files in os.walk(directory) for f in files]
|
||||||
|
|
||||||
|
|
||||||
def fetch_folders(root_dir, filter_func=None):
|
def fetch_folders(root_dir, filter_func=None):
|
||||||
folders = []
|
folders = []
|
||||||
for root, dirs, _ in os.walk(root_dir):
|
for root, dirs, _ in os.walk(root_dir):
|
||||||
@@ -26,39 +25,39 @@ def fetch_folders(root_dir, filter_func=None):
|
|||||||
folders.append(folder_path)
|
folders.append(folder_path)
|
||||||
return folders
|
return folders
|
||||||
|
|
||||||
def save_h5(file_path: str, tensor_group: Dict[str, List[Tensor]]):
|
def save_h5(file_path: str, file_name: str, tensor_group: Dict[str, List[Tensor]]):
|
||||||
os.makedirs(os.path.dirname(file_path), exist_ok=True)
|
os.makedirs(file_path, exist_ok=True)
|
||||||
with h5py.File(file_path, 'w') as f:
|
full_file_path = os.path.join(file_path, f"{file_name}.h5")
|
||||||
|
with h5py.File(full_file_path, 'w') as f:
|
||||||
for key, tensors in tensor_group.items():
|
for key, tensors in tensor_group.items():
|
||||||
grp = f.create_group(key)
|
grp = f.create_group(key)
|
||||||
grp.attrs['num_tensors'] = len(tensors)
|
|
||||||
|
|
||||||
for idx, tensor in enumerate(tensors):
|
for idx, tensor in enumerate(tensors):
|
||||||
arr = tensor.cpu().numpy()
|
arr = tensor.cpu().numpy()
|
||||||
dset = grp.create_dataset(
|
grp.create_dataset(f'data_{idx}', data=arr)
|
||||||
f'data_{idx}',
|
|
||||||
data=arr
|
|
||||||
)
|
|
||||||
dset.attrs['numel'] = tensor.numel()
|
|
||||||
|
|
||||||
def load_h5(file_path: str) -> Tuple[Dict[str, List[Tensor]], int]:
|
def load_h5(file_path: str, share_memory=True) -> Dict[str, List[Tensor]]:
|
||||||
tensor_group: Dict[str, List[Tensor]] = {}
|
tensor_group: Dict[str, List[Tensor]] = {}
|
||||||
total_samples = 0
|
|
||||||
|
|
||||||
with h5py.File(file_path, 'r') as f:
|
root_path = Path(file_path)
|
||||||
|
h5_files = list(root_path.rglob("*.h5")) + list(root_path.rglob("*.hdf5"))
|
||||||
|
|
||||||
|
for h5_file in h5_files:
|
||||||
|
with h5py.File(h5_file, 'r') as f:
|
||||||
for key in f.keys():
|
for key in f.keys():
|
||||||
grp = f[key]
|
grp = f[key]
|
||||||
dsets = []
|
dsets = []
|
||||||
for dset_name in grp.keys():
|
for dset_name in grp.keys():
|
||||||
dset = grp[dset_name]
|
dset = grp[dset_name]
|
||||||
dsets.append(torch.from_numpy(dset[:]).share_memory_())
|
tensor = torch.from_numpy(dset[:])
|
||||||
total_samples += dset.attrs.get('numel', np.prod(dset.shape))
|
if share_memory:
|
||||||
tensor_group[key] = dsets
|
tensor = tensor.share_memory_()
|
||||||
|
dsets.append(tensor)
|
||||||
|
|
||||||
num_keys = max(len(tensor_group), 1)
|
if tensor_group.get(key) is None:
|
||||||
sample_per_key = total_samples // num_keys
|
tensor_group[key] = []
|
||||||
|
tensor_group[key].extend(dsets)
|
||||||
|
|
||||||
return tensor_group, sample_per_key
|
return tensor_group
|
||||||
|
|
||||||
def comprehensive_normalization(text):
|
def comprehensive_normalization(text):
|
||||||
replacements = {
|
replacements = {
|
||||||
@@ -106,10 +105,9 @@ def dump_files(
|
|||||||
):
|
):
|
||||||
|
|
||||||
for file_path in files:
|
for file_path in files:
|
||||||
out_file_name = os.path.basename(file_path).replace(".jsonl", ".h5")
|
os.makedirs(base_out_dir, exist_ok=True)
|
||||||
out_file_path = os.path.join(base_out_dir, out_file_name)
|
|
||||||
file_name = os.path.basename(file_path)
|
file_name = os.path.basename(file_path)
|
||||||
os.makedirs(os.path.dirname(out_file_path), exist_ok=True)
|
out_file_name = file_name.split(".")[0]
|
||||||
|
|
||||||
arrows: List[Dict[str, Tensor]] = []
|
arrows: List[Dict[str, Tensor]] = []
|
||||||
|
|
||||||
@@ -137,7 +135,7 @@ def dump_files(
|
|||||||
|
|
||||||
output_package[key] = sequence
|
output_package[key] = sequence
|
||||||
|
|
||||||
save_h5(out_file_path, output_package)
|
save_h5(base_out_dir, out_file_name, output_package)
|
||||||
|
|
||||||
|
|
||||||
def get_pt_processor(tokenizer: BpeTokenizer):
|
def get_pt_processor(tokenizer: BpeTokenizer):
|
||||||
|
|||||||
Reference in New Issue
Block a user