docs: async def _run(self, data_id, script):
          """Execute the script and save results."""
          self.data_id = data_id
  
          # Fetch data instance to get any executor requirements.
          self.process = PROCESS
          requirements = self.process['requirements']
          self.requirements = requirements.get('executor', {}).get(self.name, {})  # pylint: disable=no-member
          self.resources = requirements.get('resources', {})
  
          logger.debug("Preparing output files for Data with id {}".format(data_id))
          os.chdir(EXECUTOR_SETTINGS['DATA_DIR'])
          try:
              log_file = self._create_file('stdout.txt')
              json_file = self._create_file('jsonout.txt')
          except FileExistsError:
              logger.error("Stdout or jsonout out file already exists.")
              # Looks like executor was already ran for this Data object,
              # so don't raise the error to prevent setting status to error.
              await self._send_manager_command(ExecutorProtocol.ABORT, expect_reply=False)
              return
  
          proc_pid = await self.start()
  
          await self.update_data_status(
              status=DATA_META['STATUS_PROCESSING'],
              process_pid=proc_pid
          )
  
          # Run process and handle intermediate results
          logger.info("Running program for Data with id {}".format(data_id))
          logger.debug("The program for Data with id {} is: \n{}".format(data_id, script))
          await self.run_script(script)
          spawn_processes = []
          output = {}
          process_error, process_warning, process_info = [], [], []
          process_progress, process_rc = 0, 0
  
          # read process output
          try:
              stdout = self.get_stdout()
              while True:
                  line = await stdout.readline()
                  logger.debug("Process's output: {}".format(line.strip()))
  
                  if not line:
                      break
                  line = line.decode('utf-8')
  
                  try:
                      if line.strip().startswith('run'):
                          # Save process and spawn if no errors
                          log_file.write(line)
                          log_file.flush()
  
                          for obj in iterjson(line[3:].strip()):
                              spawn_processes.append(obj)
                      elif line.strip().startswith('export'):
                          file_name = line[6:].strip()
  
                          export_folder = SETTINGS['FLOW_EXECUTOR']['UPLOAD_DIR']
                          unique_name = 'export_{}'.format(uuid.uuid4().hex)
                          export_path = os.path.join(export_folder, unique_name)
  
                          self.exported_files_mapper[self.data_id][file_name] = unique_name
  
                          shutil.move(file_name, export_path)
                      else:
                          # If JSON, save to MongoDB
                          updates = {}
                          for obj in iterjson(line):
                              for key, val in obj.items():
                                  if key.startswith('proc.'):
                                      if key == 'proc.error':
                                          process_error.append(val)
                                          if not process_rc:
                                              process_rc = 1
                                              updates['process_rc'] = process_rc
                                          updates['process_error'] = process_error
                                          updates['status'] = DATA_META['STATUS_ERROR']
                                      elif key == 'proc.warning':
                                          process_warning.append(val)
                                          updates['process_warning'] = process_warning
                                      elif key == 'proc.info':
                                          process_info.append(val)
                                          updates['process_info'] = process_info
                                      elif key == 'proc.rc':
                                          process_rc = int(val)
                                          updates['process_rc'] = process_rc
                                          if process_rc != 0:
                                              updates['status'] = DATA_META['STATUS_ERROR']
                                      elif key == 'proc.progress':
                                          process_progress = int(float(val) * 100)
                                          updates['process_progress'] = process_progress
                                  else:
                                      output[key] = val
                                      updates['output'] = output
  
                          if updates:
                              await self.update_data_status(**updates)
                              # Process meta fields are collected in listener, so we can clear them.
                              process_error, process_warning, process_info = [], [], []
  
                          if process_rc > 0:
                              log_file.close()
                              json_file.close()
                              await self._send_manager_command(ExecutorProtocol.FINISH, extra_fields={
                                  ExecutorProtocol.FINISH_PROCESS_RC: process_rc
                              })
                              return
  
                          # Debug output
                          # Not referenced in Data object
                          json_file.write(line)
                          json_file.flush()
  
                  except ValueError as ex:
                      # Ignore if not JSON
                      log_file.write(line)
                      log_file.flush()
  
          except MemoryError as ex:
              logger.error("Out of memory:\n\n{}".format(ex))
  
          except IOError as ex:
              # TODO: if ex.errno == 28: no more free space
              raise ex
          finally:
              # Store results
              log_file.close()
              json_file.close()
  
          return_code = await self.end()
  
          if process_rc < return_code:
              process_rc = return_code
  
          # send a notification to the executor listener that we're done
          finish_fields = {
              ExecutorProtocol.FINISH_PROCESS_RC: process_rc
          }
          if spawn_processes and process_rc == 0:
              finish_fields[ExecutorProtocol.FINISH_SPAWN_PROCESSES] = spawn_processes
              finish_fields[ExecutorProtocol.FINISH_EXPORTED_FILES] = self.exported_files_mapper
  
          return finish_fields