code: def __init__(
      self,
      stage_iter,
      num_tasks_per_round: int,
      max_concurrent_rounds: int = 1,
      progress_bar: Optional[ProgressBar] = None,
  ):
      self._stage_iter = stage_iter
      self._num_tasks_per_round = num_tasks_per_round
      self._max_concurrent_rounds = max_concurrent_rounds
      self._progress_bar = progress_bar
  
      self._rounds: List[List[ObjectRef]] = []
      self._task_idx = 0
  
      self._submit_round()
  
imports: import logging
  import math
  from typing import Any, Callable, Dict, List, Optional, Tuple, TypeVar, Union
  import ray
  from ray.data._internal.block_list import BlockList
  from ray.data._internal.execution.interfaces import TaskContext
  from ray.data._internal.progress_bar import ProgressBar
  from ray.data._internal.remote_fn import cached_remote_fn
  from ray.data._internal.shuffle import ShuffleOp
  from ray.data.block import Block, BlockAccessor, BlockExecStats, BlockMetadata
  from ray.data.context import DataContext
  from ray.types import ObjectRef
  from ray.util.scheduling_strategies import NodeAffinitySchedulingStrategy
label_desc: Use image node_id 1 to create a new _PipelinedStageExecutor object  with example: obj = _PipelinedStageExecutor(stage_iter, num_tasks_per_round, max_concurrent_rounds, progress_bar)