(self)
fitting using memory backend
fitting using memory backend
fitting using memory backend
def memory_fit(self):

        """
        fitting using memory backend
        """

        LOGGER.info('begin to fit homo decision tree, epoch {}, tree idx {},'
                    'running on memory backend'.format(self.epoch_idx, self.tree_idx))

        self.init_root_node_and_gh_sum()
        g, h = self.get_g_h_arr()
        self.init_memory_hist_builder(g, h, self.arr_bin_data, self.bin_num + self.use_missing) # last missing bin
        root_indices = self.init_node2index(len(self.arr_bin_data))
        self.cur_layer_node[0].inst_indices = root_indices  # root node

        tree_height = self.max_depth + 1  # non-leaf node height + 1 layer leaf

        for dep in range(tree_height):

            if dep + 1 == tree_height:
                for node in self.cur_layer_node:
                    node.is_leaf = True
                    self.tree_node.append(node)
                break

            self.sync_cur_layer_node_num(len(self.cur_layer_node), suffix=(dep, self.epoch_idx, self.tree_idx))

            node_map = self.get_node_map(self.cur_layer_node)
            node_hists = []
            for batch_id, i in enumerate(range(0, len(self.cur_layer_node), self.max_split_nodes)):

                cur_to_split = self.cur_layer_node[i:i + self.max_split_nodes]

                for node in cur_to_split:
                    if node.id in node_map:
                        hist = self.sklearn_compute_agg_hist(node.inst_indices)
                        hist_bag = HistogramBag(hist, tensor_type='array')
                        hist_bag.hid = node.id
                        hist_bag.p_hid = node.parent_nodeid
                        node_hists.append(hist_bag)

                self.sync_local_node_histogram(node_hists, suffix=(batch_id, dep, self.epoch_idx, self.tree_idx))
                node_hists = []

            split_info = self.sync_best_splits(suffix=(dep, self.epoch_idx))
            new_layer_node = self.update_tree(self.cur_layer_node, split_info)
            node2inst_idx = []

            for node in self.cur_layer_node:
                if node.is_leaf:
                    continue
                l, r = self.assign_arr_inst(node, self.arr_bin_data, node.inst_indices, missing_bin_index=self.bin_num)
                node2inst_idx.append(l)
                node2inst_idx.append(r)
            assert len(node2inst_idx) == len(new_layer_node)

            for node, indices in zip(new_layer_node, node2inst_idx):
                node.inst_indices = indices
            self.cur_layer_node = new_layer_node

        sample_indices, weights = [], []

        for node in self.tree_node:
            if node.is_leaf:
                sample_indices += list(node.inst_indices)
                weights += [node.weight] * len(node.inst_indices)
            else:
                node.bid = self.bin_split_points[node.fid][int(node.bid)]

        # post-processing of memory backend fit
        sample_id = self.sample_id_arr[sample_indices]
        self.leaf_count = {}
        for node in self.tree_node:
            if node.is_leaf:
                self.leaf_count[node.id] = len(node.inst_indices)
        LOGGER.debug('leaf count is {}'.format(self.leaf_count))
        sample_id_type = type(self.g_h.take(1)[0][0])
        self.sample_weights = session.parallelize([(sample_id_type(id_), weight) for id_, weight in zip(sample_id, weights)],
                                                  include_key=True, partition=self.data_bin.partitions)
def memory_fit ( self ) : LOGGER . info ( 'begin to fit homo decision tree, epoch {}, tree idx {},' 'running on memory backend' . format ( self . epoch_idx , self . tree_idx ) ) self . init_root_node_and_gh_sum ( ) g , h = self . get_g_h_arr ( ) self . init_memory_hist_builder ( g , h , self . arr_bin_data , self . bin_num + self . use_missing ) # last missing bin root_indices = self . init_node2index ( len ( self . arr_bin_data ) ) self . cur_layer_node [ 0 ] . inst_indices = root_indices # root node tree_height = self . max_depth + 1 # non-leaf node height + 1 layer leaf for dep in range ( tree_height ) : if dep + 1 == tree_height : for node in self . cur_layer_node : node . is_leaf = True self . tree_node . append ( node ) break self . sync_cur_layer_node_num ( len ( self . cur_layer_node ) , suffix = ( dep , self . epoch_idx , self . tree_idx ) ) node_map = self . get_node_map ( self . cur_layer_node ) node_hists = [ ] for batch_id , i in enumerate ( range ( 0 , len ( self . cur_layer_node ) , self . max_split_nodes ) ) : cur_to_split = self . cur_layer_node [ i : i + self . max_split_nodes ] for node in cur_to_split : if node . id in node_map : hist = self . sklearn_compute_agg_hist ( node . inst_indices ) hist_bag = HistogramBag ( hist , tensor_type = 'array' ) hist_bag . hid = node . id hist_bag . p_hid = node . parent_nodeid node_hists . append ( hist_bag ) self . sync_local_node_histogram ( node_hists , suffix = ( batch_id , dep , self . epoch_idx , self . tree_idx ) ) node_hists = [ ] split_info = self . sync_best_splits ( suffix = ( dep , self . epoch_idx ) ) new_layer_node = self . update_tree ( self . cur_layer_node , split_info ) node2inst_idx = [ ] for node in self . cur_layer_node : if node . is_leaf : continue l , r = self . assign_arr_inst ( node , self . arr_bin_data , node . inst_indices , missing_bin_index = self . bin_num ) node2inst_idx . append ( l ) node2inst_idx . append ( r ) assert len ( node2inst_idx ) == len ( new_layer_node ) for node , indices in zip ( new_layer_node , node2inst_idx ) : node . inst_indices = indices self . cur_layer_node = new_layer_node sample_indices , weights = [ ] , [ ] for node in self . tree_node : if node . is_leaf : sample_indices += list ( node . inst_indices ) weights += [ node . weight ] * len ( node . inst_indices ) else : node . bid = self . bin_split_points [ node . fid ] [ int ( node . bid ) ] # post-processing of memory backend fit sample_id = self . sample_id_arr [ sample_indices ] self . leaf_count = { } for node in self . tree_node : if node . is_leaf : self . leaf_count [ node . id ] = len ( node . inst_indices ) LOGGER . debug ( 'leaf count is {}' . format ( self . leaf_count ) ) sample_id_type = type ( self . g_h . take ( 1 ) [ 0 ] [ 0 ] ) self . sample_weights = session . parallelize ( [ ( sample_id_type ( id_ ) , weight ) for id_ , weight in zip ( sample_id , weights ) ] , include_key = True , partition = self . data_bin . partitions )