First draft of new featurization attempt

    self.featurizers = featurizers

    self.log_every_n = log_every_n

  def featurize(self, input_files, data_dir, shard_size=8192, worker_pool=None):

  def featurize(self, input_files, data_dir, shard_size=8192,

                num_shards_per_batch=10, worker_pool=None):

    """Featurize provided files and write to specified location."""

    log("Loading raw samples now.", self.verbosity)

    if worker_pool is None:

      worker_pool = mp.Pool(processes=1)

    metadata_rows = worker_pool.map(

    log("Spawning workers now.", self.verbosity)

    metadata_rows = []

    data_iterator = it.izip(

        it.repeat((self, shard_size, input_type, data_dir)),

        enumerate(load_data(input_files, shard_size, self.verbosity)))

    ###### TODO(rbharath): Turns out python map is terrible and exhausts the

    ###### generator as given. Solution seems to be to to manually pull out N elements

    ###### from iterator, then to map on only those N elements. BLECH. Python

    ###### should do a better job here.

    while True:

      batch_metadata = worker_pool.map(

          featurize_map_function,

        it.izip(it.repeat((self, shard_size, input_type, data_dir)),

                enumerate(load_data(input_files, shard_size))))

          itertools.islice(data_iterator, num_shards_per_batch),

          chunksize=1)

      if batch_metadata:

        metadata_rows.extend(batch_metadata)

      else:

        break

    # TODO(rbharath): This whole bit with metadata_rows is an awkward way of

    # creating a Dataset. Is there a more elegant solutions?

  else:

    raise ValueError("Unrecognized extension %s" % file_extension)

def load_data(input_files, shard_size=None):

def load_data(input_files, shard_size=None, verbosity=None):

  """Loads data from disk.

  For CSV files, supports sharded loading for large files.

  """

  if not len(input_files):

    return []

    return

  input_type = get_input_type(input_files[0])

  if input_type == "sdf":

    if shard_size is not None:

      raise ValueError("shard_size must be None for sdf input.")

    return load_sdf_files(input_files)

    for value in load_sdf_files(input_files):

      yield value

  elif input_type == "csv":

    return load_csv_files(input_files, shard_size)

    for value in load_csv_files(input_files, shard_size, verbosity=verbosity):

      yield value

  elif input_type == "pandas-pickle":

    return [load_pickle_from_disk(input_file) for input_file in input_files]

    for input_file in input_files:

      yield load_pickle_from_disk(input_file)

def load_sdf_files(input_files):

  """Load SDF file into dataframe."""

    dataframes.append(pd.concat([mol_df, raw_df], axis=1, join='inner'))

  return dataframes

def load_csv_files(filenames, shard_size=None):

def load_csv_files(filenames, shard_size=None, verbosity=None):

  """Load data as pandas dataframe."""

  # First line of user-specified CSV *must* be header.

  shard_num = 1

  for filename in filenames:

    if shard_size is None:

      yield pd.read_csv(filename)

    else:

      log("About to start loading CSV from %s" % filename, verbosity)

      for df in pd.read_csv(filename, chunksize=shard_size):

        log("Loading shard %d of size %s." % (shard_num, str(shard_size)),

            verbosity)

        df = df.replace(np.nan, str(""), regex=True)

        shard_num += 1

        yield df

def load_from_disk(filename):