Return top-k most likely predictions for each observation in
        ``dataset``. Predictions are returned as an SFrame with three columns:
        `row_id`, `class`, and `probability`.

        Parameters
        ----------
        dataset : SFrame
            Dataset of new observations. Must include the features used for
            model training, but does not require a target column. Additional
            columns are ignored.

        max_neighbors : int, optional
            Maximum number of neighbors to consider for each point.

        radius : float, optional
            Maximum distance from each point to a neighbor in the reference
            dataset.

        k : int, optional
            Number of classes to return for each input example.

        Returns
        -------
        out : SFrame

        See Also
        ----------
        create, classify, predict

        Notes
        -----
        - If the 'radius' parameter is small, it is possible that a query point
          has no neighbors in the training dataset. In this case, the query is
          dropped from the SFrame output by this method. If all queries have no
          neighbors, then the result is an empty SFrame. If the target column in
          the training dataset has missing values, these predictions will be
          ambiguous.

        - Ties between predicted classes are broken randomly.

        Examples
        --------
        >>> sf_train = turicreate.SFrame({'species': ['cat', 'dog', 'fossa', 'dog'],
        ...                             'height': [9, 25, 20, 23],
        ...                             'weight': [13, 28, 33, 22]})
        ...
        >>> sf_new = turicreate.SFrame({'height': [26, 19],
        ...                           'weight': [25, 35]})
        ...
        >>> m = turicreate.nearest_neighbor_classifier.create(sf_train, target='species')
        >>> ystar = m.predict_topk(sf_new, max_neighbors=2)
        >>> print ystar
        +--------+-------+-------------+
        | row_id | class | probability |
        +--------+-------+-------------+
        |   0    |  dog  |     1.0     |
        |   1    | fossa |     0.5     |
        |   1    |  dog  |     0.5     |
        +--------+-------+-------------+

def predict_topk(self, dataset,  max_neighbors=10, radius=None, k=3,
                     verbose=False):
        """
        Return top-k most likely predictions for each observation in
        ``dataset``. Predictions are returned as an SFrame with three columns:
        `row_id`, `class`, and `probability`.

        Parameters
        ----------
        dataset : SFrame
            Dataset of new observations. Must include the features used for
            model training, but does not require a target column. Additional
            columns are ignored.

        max_neighbors : int, optional
            Maximum number of neighbors to consider for each point.

        radius : float, optional
            Maximum distance from each point to a neighbor in the reference
            dataset.

        k : int, optional
            Number of classes to return for each input example.

        Returns
        -------
        out : SFrame

        See Also
        ----------
        create, classify, predict

        Notes
        -----
        - If the 'radius' parameter is small, it is possible that a query point
          has no neighbors in the training dataset. In this case, the query is
          dropped from the SFrame output by this method. If all queries have no
          neighbors, then the result is an empty SFrame. If the target column in
          the training dataset has missing values, these predictions will be
          ambiguous.

        - Ties between predicted classes are broken randomly.

        Examples
        --------
        >>> sf_train = turicreate.SFrame({'species': ['cat', 'dog', 'fossa', 'dog'],
        ...                             'height': [9, 25, 20, 23],
        ...                             'weight': [13, 28, 33, 22]})
        ...
        >>> sf_new = turicreate.SFrame({'height': [26, 19],
        ...                           'weight': [25, 35]})
        ...
        >>> m = turicreate.nearest_neighbor_classifier.create(sf_train, target='species')
        >>> ystar = m.predict_topk(sf_new, max_neighbors=2)
        >>> print ystar
        +--------+-------+-------------+
        | row_id | class | probability |
        +--------+-------+-------------+
        |   0    |  dog  |     1.0     |
        |   1    | fossa |     0.5     |
        |   1    |  dog  |     0.5     |
        +--------+-------+-------------+
        """

        ## Validate the number of results to return. Note that the
        #  'max_neighbors' and 'radius' parameters are validated by the nearest
        #  neighbor model's query method.
        if not isinstance(k, int) or k < 1:
            raise TypeError("The number of results to return for each point, " +
                            "'k', must be an integer greater than 0.")


        ## Validate the query dataset.
        _raise_error_if_not_sframe(dataset, "dataset")
        _raise_error_if_sframe_empty(dataset, "dataset")

        ## Validate neighborhood parameters 'max_neighbors'.
        # - NOTE: when the parameter name is changed in nearest neighbors, the
        #   query call will do this itself, and this block can be removed.
        if max_neighbors is not None:
            if not isinstance(max_neighbors, int):
                raise ValueError("Input 'max_neighbors' must be an integer.")

            if max_neighbors <= 0:
                raise ValueError("Input 'max_neighbors' must be larger than 0.")


        ## Find the nearest neighbors for each query and count the number of
        #  votes for each class.
        knn = self._knn_model.query(dataset, k=max_neighbors, radius=radius,
                                    verbose=verbose)

        ## If there are *no* results for *any* query make an empty SFrame.
        if knn.num_rows() == 0:
            ystar = _tc.SFrame({'row_id': [], 'class': [], 'probability': []})
            ystar['row_id'] = ystar['row_id'].astype(int)
            ystar['class'] = ystar['class'].astype(str)


        else:
            ## Find the classes with the top-k vote totals
            grp = knn.groupby(['query_label', 'reference_label'],
                              _tc.aggregate.COUNT)

            ystar = grp.unstack(column_names=['reference_label', 'Count'],
                                new_column_name='votes')

            ystar['topk'] = ystar['votes'].apply(
                lambda x: _sort_topk_votes(x, k))
            ystar['total_votes'] = ystar['votes'].apply(
                lambda x: sum(x.values()))

            ## Re-stack, unpack, and rename the results
            ystar = ystar.stack('topk', new_column_name='topk')
            ystar = ystar.unpack('topk')
            ystar.rename({'topk.class': 'class', 'query_label': 'row_id'}, inplace=True)
            ystar['probability'] = ystar['topk.votes'] / ystar['total_votes']
            ystar = ystar[['row_id', 'class', 'probability']]

        return ystar