<SYSTEM_TASK:>
Solve the following problem using Python, implementing the functions described below, one line at a time
<END_TASK>
<USER_TASK:>
Description:
 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)