class_docstr: A streaming Dataset executor.
  
  This implementation executes Dataset DAGs in a fully streamed way. It runs
  by setting up the operator topology, and then routing blocks through operators in
  a way that maximizes throughput under resource constraints.
class_docstr_tok: ["A","streaming","Dataset","executor",".","This","implementation","executes","Dataset","DAGs","in","a","fully","streamed","way",".","It","runs","by","setting","up","the","operator","topology",",","and","then","routing","blocks","through","operators","in","a","way","that","maximizes","throughput","under","resource","constraints","."]
docstr: Executes the DAG using a streaming execution strategy.
  
  We take an event-loop approach to scheduling. We block on the next scheduling
  event using `ray.wait`, updating operator state and dispatching new tasks.
docstr_tok: ["Executes","the","DAG","using","a","streaming","execution","strategy",".","We","take","an","event-loop","approach","to","scheduling",".","We","block","on","the","next","scheduling","event","using","`","ray.wait","`",",","updating","operator","state","and","dispatching","new","tasks","."]
code: def execute(
      self,
      dag: PhysicalOperator,
      initial_stats: Optional[DatasetStats] = None,
  ) -> Iterator[RefBundle]:
      """Executes the DAG using a streaming execution strategy.
  
      We take an event-loop approach to scheduling. We block on the next scheduling
      event using `ray.wait`, updating operator state and dispatching new tasks.
      """
  
      self._initial_stats = initial_stats
      self._start_time = time.perf_counter()
  
      if not isinstance(dag, InputDataBuffer):
          logger.get_logger().info("Executing DAG %s", dag)
          logger.get_logger().info(
              "Execution config: %s", self._options
          )
          if not self._options.verbose_progress:
              logger.get_logger().info(
                  "Tip: For detailed progress reporting, run "
                  "`ray.data.DataContext.get_current()."
                  "execution_options.verbose_progress = True`"
              )
  
      # Setup the streaming DAG topology and start the runner thread.
      self._topology, _ = build_streaming_topology(dag, self._options)
      self._backpressure_policies = get_backpressure_policies(
          self._topology
      )
  
      self._has_op_completed = {op: False for op in self._topology}
  
      if not isinstance(dag, InputDataBuffer):
          # Note: DAG must be initialized in order to query num_outputs_total.
          self._global_info = ProgressBar(
              "Running", dag.num_outputs_total()
          )
  
      self._output_node: OpState = self._topology[dag]
      StatsManager.register_dataset_to_stats_actor(
          self._dataset_tag,
          self._get_operator_tags(),
      )
      self.start()
      self._execution_started = True
  
      class StreamIterator(OutputIterator):
          def __init__(self, outer: Executor):
              self._outer = outer
  
          def get_next(
              self, output_split_idx: Optional[int] = None
          ) -> RefBundle:
              try:
                  item = self._outer._output_node.get_output_blocking(
                      output_split_idx
                  )
                  if self._outer._global_info:
                      self._outer._global_info.update(
                          1, dag._estimated_output_blocks
                      )
                  return item
              # Needs to be BaseException to catch KeyboardInterrupt. Otherwise we
              # can leave dangling progress bars by skipping shutdown.
              except BaseException as e:
                  self._outer.shutdown(isinstance(e, StopIteration))
                  raise
  
          def __del__(self):
              self._outer.shutdown()
  
      return StreamIterator(self)
  
imports: import os
  import threading
  import time
  import uuid
  from typing import Dict, Iterator, List, Optional
  from ray.data._internal.dataset_logger import DatasetLogger
  from ray.data._internal.execution.autoscaling_requester import get_or_create_autoscaling_requester_actor
  from ray.data._internal.execution.backpressure_policy import BackpressurePolicy, get_backpressure_policies
  from ray.data._internal.execution.interfaces import ExecutionOptions, ExecutionResources, Executor, OutputIterator, PhysicalOperator, RefBundle
  from ray.data._internal.execution.operators.input_data_buffer import InputDataBuffer
  from ray.data._internal.execution.streaming_executor_state import DEFAULT_OBJECT_STORE_MEMORY_LIMIT_FRACTION, AutoscalingState, OpState, Topology, TopologyResourceUsage, build_streaming_topology, process_completed_tasks, select_operator_to_run, update_operator_states
  from ray.data._internal.progress_bar import ProgressBar
  from ray.data._internal.stats import DatasetStats, StatsManager
  from ray.data._internal.util import cluster_resources
  from ray.data.context import DataContext
label_desc: Use image node_id 2 for calling the StreamingExecutor obj's underlying member method code with example usage: obj.execute(dag, initial_stats) and returns: StreamIterator, item