Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

debug for gpu rank for analyser #329

Merged
merged 4 commits into from
Jun 25, 2024
Merged

debug for gpu rank for analyser #329

Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
71158fb
debug for gpu rank for analyser
BeachWang Jun 11, 2024
054ab58
after pre-commit
BeachWang Jun 11, 2024
2b9cdfc
spec_numprocs -> num_proc
BeachWang Jun 24, 2024
8febfbe
save fix
BeachWang Jun 25, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 11 additions & 1 deletion data_juicer/core/analyser.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,12 +2,14 @@

from loguru import logger

from data_juicer import use_cuda
from data_juicer.analysis import ColumnWiseAnalysis, OverallAnalysis
from data_juicer.config import init_configs
from data_juicer.format import load_formatter
from data_juicer.ops import Filter, load_ops
from data_juicer.utils import cache_utils
from data_juicer.utils.constant import Fields
from data_juicer.utils.process_utils import calculate_np

from .data import add_same_content_to_new_column
from .exporter import Exporter
Expand Down Expand Up @@ -89,6 +91,13 @@ def run(self, load_data_np=None, skip_export=False):
stats_collected = False
for op_cfg, op in zip(self.cfg.process, self.ops):
op_name = list(op_cfg.keys())[0]
with_rank = use_cuda() and op._accelerator == 'cuda'
if op.spec_numprocs != 0:
BeachWang marked this conversation as resolved.
Show resolved Hide resolved
op_proc = op.spec_numprocs
logger.info(f'Op [{op_name}] running with sepcified '
f'number of procs:{op.spec_numprocs}')
else:
op_proc = calculate_np(self.cfg.np, op, op_name)
if isinstance(op, Filter):
if Fields.stats not in dataset.features:
# only add stats when calling filter op
Expand All @@ -100,7 +109,8 @@ def run(self, load_data_np=None, skip_export=False):
num_proc=self.cfg.np,
desc='Adding new column for stats')
dataset = dataset.map(op.compute_stats,
num_proc=self.cfg.np,
num_proc=op_proc,
with_rank=with_rank,
desc=op_name + '_compute_stats')
stats_collected = True
if not stats_collected:
Expand Down
Loading