def METHOD_NAME(self, data_instances=None):
    LOGGER.info("Start predict ...")
    res_dict = self.aggregator.aggregate_model(suffix=('predict', ))
    cluster_result = res_dict.mapValues(lambda v: np.argmin(v))
    cluster_dist_result = res_dict.mapValues(lambda v: min(v))
    self.aggregator.broadcast_model(cluster_result, suffix=('predict', ))
    res_dict_dbi = self.aggregator.aggregate_model(suffix=('predict_dbi', ))
    self.aggregator.broadcast_model(cluster_result, suffix=('predict_dbi', ))
    dist_cluster_table = res_dict.join(cluster_result, lambda v1, v2: [v1, v2])
    dist_cluster_table_dbi = res_dict_dbi.join(cluster_result, lambda v1, v2: [v1, v2])
    dist_table = self.cal_ave_dist(dist_cluster_table, cluster_result)  # ave dist in each cluster
    dist_table_dbi = self.cal_ave_dist(dist_cluster_table_dbi, cluster_result)
    cluster_dist = self.aggregator.aggregate_model(suffix=('predict_cluster_dist', ))
    dist_cluster_table_out = cluster_result.join(cluster_dist_result, lambda v1, v2: [int(v1), float(v2)])
    cluster_max_radius = dist_cluster_table_out.applyPartitions(self.max_radius).reduce(self.get_max_radius)
    result = []
    for i in range(len(dist_table)):
        c_key = dist_table[i][0]
        result.append(tuple(
            [int(c_key),
             Instance(features=[dist_table[i][1], dist_table_dbi[i][2],
                                cluster_max_radius[c_key], list(cluster_dist._weights)])]))
    predict_result1 = session.parallelize(result, partition=res_dict.partitions, include_key=True)
    predict_result2 = dist_cluster_table_out.mapValues(lambda x: Instance(features=x))
    return predict_result1, predict_result2