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 _get_state_dict(self, state):
      last_op, last_state = list(self._topology.items())[-1]
      return {
          "state": state,
          "progress": last_state.num_completed_tasks,
          "total": last_op.num_outputs_total(),
          "end_time": time.time() if state != "RUNNING" else None,
          "operators": {
              f"{op.name}{i}": {
                  "progress": op_state.num_completed_tasks,
                  "total": op.num_outputs_total(),
                  "state": state,
              }
              for i, (op, op_state) in enumerate(self._topology.items())
          },
      }
  
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 13 for calling the StreamingExecutor obj's underlying member method code with example usage: obj._get_state_dict(state) and returns: dict