code: def fix_paths(job):
      
      tmp_files = []
      args = []
      for x in job.args():
          if isinstance(x, luigi.contrib.hdfs.HdfsTarget):  # input/output
              if x.exists() or not job.atomic_output():  # input
                  args.append(x.path)
              else:  # output
                  x_path_no_slash = x.path[:-1] if x.path[-1] == '/' else x.path
                  y = luigi.contrib.hdfs.HdfsTarget(x_path_no_slash + '-luigi-tmp-%09d' % random.randrange(0, 1e10))
                  tmp_files.append((y, x_path_no_slash))
                  logger.info('Using temp path: %s for path %s', y.path, x.path)
                  args.append(y.path)
          else:
              try:
                  # hopefully the target has a path to use
                  args.append(x.path)
              except AttributeError:
                  # if there's no path then hope converting it to a string will work
                  args.append(str(x))
  
      return (tmp_files, args)
docs: Coerce input arguments to use temporary files when used for output.
  
      Return a list of temporary file pairs (tmpfile, destination path) and
      a list of arguments.
  
      Converts each HdfsTarget to a string for the path.