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","."]
code: def __init__(
      self,
      options: ExecutionOptions,
      dataset_tag: str = "unknown_dataset",
  ):
      self._start_time: Optional[float] = None
      self._initial_stats: Optional[DatasetStats] = None
      self._final_stats: Optional[DatasetStats] = None
      self._global_info: Optional[ProgressBar] = None
  
      self._execution_id = uuid.uuid4().hex
      self._autoscaling_state = AutoscalingState()
  
      # The executor can be shutdown while still running.
      self._shutdown_lock = threading.RLock()
      self._execution_started = False
      self._shutdown = False
  
      # Internal execution state shared across thread boundaries. We run the control
      # loop on a separate thread so that it doesn't become stalled between
      # generator `yield`s.
      self._topology: Optional[Topology] = None
      self._output_node: Optional[OpState] = None
      self._backpressure_policies: List[BackpressurePolicy] = []
  
      self._dataset_tag = dataset_tag
      # Stores if an operator is completed,
      # used for marking when an op has just completed.
      self._has_op_completed: Optional[
          Dict[PhysicalOperator, bool]
      ] = None
      self._max_errored_blocks = (
          DataContext.get_current().max_errored_blocks
      )
      self._num_errored_blocks = 0
  
      self._last_debug_log_time = 0
  
      Executor.__init__(self, options)
      thread_name = f"StreamingExecutor-{self._execution_id}"
      threading.Thread.__init__(self, daemon=True, name=thread_name)
  
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 1 to create a new StreamingExecutor object from inherited base classes: Executor, threading with example: obj = StreamingExecutor(options, dataset_tag)