code: def start(self):
          
  
          def cb():
              time_ = time.time()
              log.debug('Step {}'.format(time_))
  
              # run a step on all streams
              for d in self._dstreams:
                  d._step(time_)
  
          self._pcb = PeriodicCallback(cb, self.batch_duration * 1000.0)
          self._pcb.start()
          self._on_stop_cb.append(self._pcb.stop)
          StreamingContext._activeContext = self
docs: Start processing streams.