code: def split_input(cls, job_config):
      
      reader_params = job_config.input_reader_params
      bucket = reader_params[cls.BUCKET_NAME_PARAM]
      filenames = reader_params[cls.OBJECT_NAMES_PARAM]
      delimiter = reader_params.get(cls.DELIMITER_PARAM)
      account_id = reader_params.get(cls._ACCOUNT_ID_PARAM)
      buffer_size = reader_params.get(cls.BUFFER_SIZE_PARAM)
      path_filter = reader_params.get(cls.PATH_FILTER_PARAM)
  
      # Gather the complete list of files (expanding wildcards)
      all_filenames = []
      for filename in filenames:
        if filename.endswith("*"):
          all_filenames.extend(
              [file_stat.filename for file_stat in cloudstorage.listbucket(
                  "/" + bucket + "/" + filename[:-1], delimiter=delimiter,
                  _account_id=account_id)])
        else:
          all_filenames.append("/%s/%s" % (bucket, filename))
  
      # Split into shards
      readers = []
      for shard in range(0, job_config.shard_count):
        shard_filenames = all_filenames[shard::job_config.shard_count]
        if shard_filenames:
          readers.append(cls(
              shard_filenames, buffer_size=buffer_size, _account_id=account_id,
              delimiter=delimiter, path_filter=path_filter))
      return readers
docs: Returns a list of input readers.
  
      An equal number of input files are assigned to each shard (+/- 1). If there
      are fewer files than shards, fewer than the requested number of shards will
      be used. Input files are currently never split (although for some formats
      could be and may be split in a future implementation).
  
      Args:
        job_config: map_job.JobConfig
  
      Returns:
        A list of InputReaders. None when no input data can be found.