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

[RayExecutor]Check unsupported op before processing #609

Closed
wants to merge 1 commit into from
+9 −2
Closed

[RayExecutor]Check unsupported op before processing #609

Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
11 changes: 9 additions & 2 deletions data_juicer/core/ray_executor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@

from data_juicer.config import init_configs
from data_juicer.core.ray_data import RayDataset
from data_juicer.ops import load_ops
from data_juicer.ops import Aggregator, Grouper, Selector, load_ops
from data_juicer.ops.op_fusion import fuse_operators

from .adapter import Adapter
Expand Down Expand Up @@ -53,7 +53,7 @@ def __init__(self, cfg=None):
self.adapter = Adapter(self.cfg)

# init ray
logger.info('Initing Ray ...')
logger.info('Initializing Ray ...')
ray.init(self.cfg.ray_address)
self.tmp_dir = os.path.join(self.work_dir, '.tmp',
ray.get_runtime_context().get_job_id())
Expand Down Expand Up @@ -85,6 +85,13 @@ def run(self, load_data_np=None):
logger.info('Preparing process operators...')
ops = load_ops(self.cfg.process)

# check if the ops contain unsupported operators
# raise error before processing
for op in ops:
if isinstance(op, (Selector, Grouper, Aggregator)):
raise ValueError(
f'Operator {op} is not supported in RayExecutor')

if self.cfg.op_fusion:
probe_res = None
if self.cfg.fusion_strategy == 'probe':
Expand Down
Loading