(self, data_inst, sample_ids=None)
Stratified sample method, a line's occur probability is decide by fractions
            Input should be Table, every line should be an instance object with label
            To use this method, a list of ratio should be give, and the list length
                equals to the number of distinct labels
            support down sample and up sample
                if use down sample: should give a list of (category, ratio), where ratio is between [0, 1]
                otherwise: should give a list (category, ratio), where the float ratio should no less than 1.0


        Parameters
        ----------
        data_inst : Table
            The input data

        sample_ids : None or list
            if None, will sample data from the class instance's parameters,
            otherwise, it will be sample transform process, which means use the samples_ids the generate data

        Returns
        -------
        new_data_inst: Table
            the output sample data, sample format with input

        sample_ids: list, return only if sample_ids is None
Stratified sample method, a line's occur probability is decide by fractions
            Input should be Table, every line should be an instance object with label
            To use this method, a list of ratio should be give, and the list length
                equals to the number of distinct labels
            support down sample and up sample
                if use down sample: should give a list of (category, ratio), where ratio is between [0, 1]
                otherwise: should give a list (category, ratio), where the float ratio should no less than 1.0
Stratified sample method a line s occur probability is decide by fractions Input should be Table every line should be an instance object with label To use this method a list of ratio should be give and the list length equals to the number of distinct labels support down sample and up sample if use down sample : should give a list of ( category ratio ) where ratio is between [ 0 1 ] otherwise : should give a list ( category ratio ) where the float ratio should no less than 1 . 0
def __sample(self, data_inst, sample_ids=None):
        """
        Stratified sample method, a line's occur probability is decide by fractions
            Input should be Table, every line should be an instance object with label
            To use this method, a list of ratio should be give, and the list length
                equals to the number of distinct labels
            support down sample and up sample
                if use down sample: should give a list of (category, ratio), where ratio is between [0, 1]
                otherwise: should give a list (category, ratio), where the float ratio should no less than 1.0


        Parameters
        ----------
        data_inst : Table
            The input data

        sample_ids : None or list
            if None, will sample data from the class instance's parameters,
            otherwise, it will be sample transform process, which means use the samples_ids the generate data

        Returns
        -------
        new_data_inst: Table
            the output sample data, sample format with input

        sample_ids: list, return only if sample_ids is None


        """

        LOGGER.info("start to run stratified sampling")
        return_sample_ids = False
        if self.method == "downsample":
            if sample_ids is None:
                idset = [[] for i in range(len(self.fractions))]
                for label, fraction in self.fractions:
                    if fraction < 0 or fraction > 1:
                        raise ValueError("sapmle fractions should be a numeric number between 0 and 1inclusive")

                return_sample_ids = True
                for key, inst in data_inst.collect():
                    label = inst.label
                    if label not in self.label_mapping:
                        raise ValueError("label not specify sample rate! check it please")
                    idset[self.label_mapping[label]].append(key)

                sample_ids = []

                callback_sample_metrics = []
                callback_original_metrics = []

                for i in range(len(idset)):
                    label_name = self.labels[i]
                    callback_original_metrics.append(Metric(label_name, len(idset[i])))

                    if idset[i]:
                        sample_num = max(1, int(self.fractions[i][1] * len(idset[i])))

                        _sample_ids = resample(idset[i],
                                               replace=False,
                                               n_samples=sample_num,
                                               random_state=self.random_state)

                        sample_ids.extend(_sample_ids)

                        callback_sample_metrics.append(Metric(label_name, len(_sample_ids)))
                    else:
                        callback_sample_metrics.append(Metric(label_name, 0))

                random.shuffle(sample_ids)

                callback(self.tracker, "stratified", callback_sample_metrics, callback_original_metrics, self._summary_buf)

            sample_dtable = session.parallelize(zip(sample_ids, range(len(sample_ids))),
                                                include_key=True,
                                                partition=data_inst.partitions)
            new_data_inst = data_inst.join(sample_dtable, lambda v1, v2: v1)

            if return_sample_ids:
                return new_data_inst, sample_ids
            else:
                return new_data_inst

        elif self.method == "upsample":
            data_set = list(data_inst.collect())
            ids = [key for (key, inst) in data_set]
            id_maps = dict(zip(ids, range(len(ids))))

            return_sample_ids = False

            if sample_ids is None:
                idset = [[] for i in range(len(self.fractions))]
                for label, fraction in self.fractions:
                    if fraction <= 0:
                        raise ValueError("sapmle fractions should be a numeric number greater than 0")

                for key, inst in data_set:
                    label = inst.label
                    if label not in self.label_mapping:
                        raise ValueError("label not specify sample rate! check it please")
                    idset[self.label_mapping[label]].append(key)

                return_sample_ids = True

                sample_ids = []
                callback_sample_metrics = []
                callback_original_metrics = []

                for i in range(len(idset)):
                    label_name = self.labels[i]
                    callback_original_metrics.append(Metric(label_name, len(idset[i])))

                    if idset[i]:
                        sample_num = max(1, int(self.fractions[i][1] * len(idset[i])))

                        _sample_ids = resample(idset[i],
                                               replace=True,
                                               n_samples=sample_num,
                                               random_state=self.random_state)

                        sample_ids.extend(_sample_ids)

                        callback_sample_metrics.append(Metric(label_name, len(_sample_ids)))
                    else:
                        callback_sample_metrics.append(Metric(label_name, 0))

                random.shuffle(sample_ids)

                callback(self.tracker, "stratified", callback_sample_metrics, callback_original_metrics, self._summary_buf)

            new_data = []
            for i in range(len(sample_ids)):
                index = id_maps[sample_ids[i]]
                new_data.append((i, data_set[index][1]))

            new_data_inst = session.parallelize(new_data,
                                                include_key=True,
                                                partition=data_inst.partitions)

            if return_sample_ids:
                return new_data_inst, sample_ids
            else:
                return new_data_inst

        else:
            raise ValueError("Stratified sampler not support method {} yet".format(self.method))
def __sample ( self , data_inst , sample_ids = None ) : LOGGER . info ( "start to run stratified sampling" ) return_sample_ids = False if self . method == "downsample" : if sample_ids is None : idset = [ [ ] for i in range ( len ( self . fractions ) ) ] for label , fraction in self . fractions : if fraction < 0 or fraction > 1 : raise ValueError ( "sapmle fractions should be a numeric number between 0 and 1inclusive" ) return_sample_ids = True for key , inst in data_inst . collect ( ) : label = inst . label if label not in self . label_mapping : raise ValueError ( "label not specify sample rate! check it please" ) idset [ self . label_mapping [ label ] ] . append ( key ) sample_ids = [ ] callback_sample_metrics = [ ] callback_original_metrics = [ ] for i in range ( len ( idset ) ) : label_name = self . labels [ i ] callback_original_metrics . append ( Metric ( label_name , len ( idset [ i ] ) ) ) if idset [ i ] : sample_num = max ( 1 , int ( self . fractions [ i ] [ 1 ] * len ( idset [ i ] ) ) ) _sample_ids = resample ( idset [ i ] , replace = False , n_samples = sample_num , random_state = self . random_state ) sample_ids . extend ( _sample_ids ) callback_sample_metrics . append ( Metric ( label_name , len ( _sample_ids ) ) ) else : callback_sample_metrics . append ( Metric ( label_name , 0 ) ) random . shuffle ( sample_ids ) callback ( self . tracker , "stratified" , callback_sample_metrics , callback_original_metrics , self . _summary_buf ) sample_dtable = session . parallelize ( zip ( sample_ids , range ( len ( sample_ids ) ) ) , include_key = True , partition = data_inst . partitions ) new_data_inst = data_inst . join ( sample_dtable , lambda v1 , v2 : v1 ) if return_sample_ids : return new_data_inst , sample_ids else : return new_data_inst elif self . method == "upsample" : data_set = list ( data_inst . collect ( ) ) ids = [ key for ( key , inst ) in data_set ] id_maps = dict ( zip ( ids , range ( len ( ids ) ) ) ) return_sample_ids = False if sample_ids is None : idset = [ [ ] for i in range ( len ( self . fractions ) ) ] for label , fraction in self . fractions : if fraction <= 0 : raise ValueError ( "sapmle fractions should be a numeric number greater than 0" ) for key , inst in data_set : label = inst . label if label not in self . label_mapping : raise ValueError ( "label not specify sample rate! check it please" ) idset [ self . label_mapping [ label ] ] . append ( key ) return_sample_ids = True sample_ids = [ ] callback_sample_metrics = [ ] callback_original_metrics = [ ] for i in range ( len ( idset ) ) : label_name = self . labels [ i ] callback_original_metrics . append ( Metric ( label_name , len ( idset [ i ] ) ) ) if idset [ i ] : sample_num = max ( 1 , int ( self . fractions [ i ] [ 1 ] * len ( idset [ i ] ) ) ) _sample_ids = resample ( idset [ i ] , replace = True , n_samples = sample_num , random_state = self . random_state ) sample_ids . extend ( _sample_ids ) callback_sample_metrics . append ( Metric ( label_name , len ( _sample_ids ) ) ) else : callback_sample_metrics . append ( Metric ( label_name , 0 ) ) random . shuffle ( sample_ids ) callback ( self . tracker , "stratified" , callback_sample_metrics , callback_original_metrics , self . _summary_buf ) new_data = [ ] for i in range ( len ( sample_ids ) ) : index = id_maps [ sample_ids [ i ] ] new_data . append ( ( i , data_set [ index ] [ 1 ] ) ) new_data_inst = session . parallelize ( new_data , include_key = True , partition = data_inst . partitions ) if return_sample_ids : return new_data_inst , sample_ids else : return new_data_inst else : raise ValueError ( "Stratified sampler not support method {} yet" . format ( self . method ) )