docs: def records(self, sc, group_by='greedy', limit=None, sample=1, seed=42, decode=None, summaries=None):
          """Retrieve the elements of a Dataset
  
          :param sc: a SparkContext object
          :param group_by: specifies a partition strategy for the objects
          :param limit: maximum number of objects to retrieve
          :param decode: an optional transformation to apply to the objects retrieved
          :param sample: percentage of results to return. Useful to return a sample
              of the dataset. This parameter is ignored when `limit` is set.
          :param seed: initialize internal state of the random number generator (42 by default).
              This is used to make the dataset sampling reproducible. It can be set to None to obtain
              different samples.
          :param summaries: an iterable containing a summary for each item in the dataset. If None,
              it will computed calling the summaries dataset.
          :return: a Spark rdd containing the elements retrieved
  
          """
          decode = decode or message_parser.parse_heka_message
          summaries = summaries or self.summaries(sc, limit)
  
          # Calculate the sample if summaries is not empty and limit is not set
          if summaries and limit is None and sample != 1:
              if sample < 0 or sample > 1:
                  raise ValueError('sample must be between 0 and 1')
              print(
                  "WARNING: THIS IS NOT A REPRESENTATIVE SAMPLE.\n"
                  "This 'sampling' is based on s3 files and is highly\n"
                  "susceptible to skew. Use only for quicker performance\n"
                  "while prototyping."
              )
              # We want this sample to be reproducible.
              # See https://bugzilla.mozilla.org/show_bug.cgi?id=1318681
              seed_state = random.getstate()
              try:
                  random.seed(seed)
                  summaries = random.sample(summaries,
                                            int(len(summaries) * sample))
              finally:
                  random.setstate(seed_state)
  
          # Obtain size in MB
          total_size = functools.reduce(lambda acc, item: acc + item['size'], summaries, 0)
          total_size_mb = total_size / float(1 << 20)
          print("fetching %.5fMB in %s files..." % (total_size_mb, len(summaries)))
  
          if group_by == 'equal_size':
              groups = _group_by_equal_size(summaries, 10*sc.defaultParallelism)
          elif group_by == 'greedy':
              groups = _group_by_size_greedy(summaries, 10*sc.defaultParallelism)
          else:
              raise Exception("group_by specification is invalid")
  
          self._compile_selection()
  
          keys = (
              sc.parallelize(groups, len(groups))
              .flatMap(lambda x: x)
              .map(lambda x: x['key'])
          )
          file_handles = keys.map(self.store.get_key)
  
          # decode(fp: file-object) -> list[dict]
          data = file_handles.flatMap(decode)
  
          return data.map(self._apply_selection)