code: def job_runner(self):
          # We recommend that you define a subclass, override this method and set up your own config
          
          outputs = luigi.task.flatten(self.output())
          for output in outputs:
              if not isinstance(output, luigi.contrib.hdfs.HdfsTarget):
                  warnings.warn("Job is using one or more non-HdfsTarget outputs" +
                                " so it will be run in local mode")
                  return LocalJobRunner()
          else:
              return DefaultHadoopJobRunner()
docs: Get the MapReduce runner for this job.
  
          If all outputs are HdfsTargets, the DefaultHadoopJobRunner will be used.
          Otherwise, the LocalJobRunner which streams all data through the local machine
          will be used (great for testing).