class_docstr: Represents an OpTask that handles Block data.
class_docstr_tok: ["Represents","an","OpTask","that","handles","Block","data","."]
docstr: Args:
      streaming_gen: The streaming generator of this task. It should yield blocks.
      output_ready_callback: The callback to call when a new RefBundle is output
          from the generator.
      task_done_callback: The callback to call when the task is done.
docstr_tok: ["Args",":","streaming_gen",":","The","streaming","generator","of","this","task",".","It","should","yield","blocks",".","output_ready_callback",":","The","callback","to","call","when","a","new","RefBundle","is","output","from","the","generator",".","task_done_callback",":","The","callback","to","call","when","the","task","is","done","."]
code: def __init__(
      self,
      task_index: int,
      streaming_gen: ObjectRefGenerator,
      output_ready_callback: Callable[[RefBundle], None],
      task_done_callback: Callable[[Optional[Exception]], None],
  ):
      """
      Args:
          streaming_gen: The streaming generator of this task. It should yield blocks.
          output_ready_callback: The callback to call when a new RefBundle is output
              from the generator.
          task_done_callback: The callback to call when the task is done.
      """
      super().__init__(task_index)
      # TODO(hchen): Right now, the streaming generator is required to yield a Block
      # and a BlockMetadata each time. We should unify task submission with an unified
      # interface. So each individual operator don't need to take care of the
      # BlockMetadata.
      self._streaming_gen = streaming_gen
      self._output_ready_callback = output_ready_callback
      self._task_done_callback = task_done_callback
  
imports: from abc import ABC, abstractmethod
  from typing import Any, Callable, Dict, List, Optional, Union
  import ray
  from .ref_bundle import RefBundle
  from ray._raylet import ObjectRefGenerator
  from ray.data._internal.execution.interfaces.execution_options import ExecutionOptions, ExecutionResources
  from ray.data._internal.execution.interfaces.op_runtime_metrics import OpRuntimeMetrics
  from ray.data._internal.logical.interfaces import Operator
  from ray.data._internal.stats import StatsDict
  from ray.data.context import DataContext
label_desc: Use image node_id 1 to create a new DataOpTask object from inherited base classes: OpTask with example: obj = DataOpTask(task_index, streaming_gen, output_ready_callback, task_done_callback)