(self, data_instances)
return intersect_ids
if self.task_version_id is not None:
            namespace = "#".join([str(self.guest_party_id), str(self.host_party_id), "mountain"])
            for k, v in enumerate(recv_ids_list):
                table_name = '_'.join([self.task_version_id, str(k)])
                self.tracker.job_tracker.save_as_table(v, table_name, namespace)
                LOGGER.info("save guest_{}'s id in name:{}, namespace:{}".format(k, table_name, namespace))
if self.task_version_id is not None:
            namespace = "#".join([str(self.guest_party_id), str(self.host_party_id), "mountain"])
            for k, v in enumerate(recv_ids_list):
                table_name = '_'.join([self.task_version_id, str(k)])
                self.tracker.job_tracker.save_as_table(v, table_name, namespace)
                LOGGER.info("save guest_{}'s id in name:{}, namespace:{}".format(k, table_name, namespace))
if self . task_version_id is not None : namespace = # . join ( [ str ( self . guest_party_id ) str ( self . host_party_id ) mountain ] ) for k v in enumerate ( recv_ids_list ) : table_name = _ . join ( [ self . task_version_id str ( k ) ] ) self . tracker . job_tracker . save_as_table ( v table_name namespace ) LOGGER . info ( save guest_ {} s id in name : {} namespace : {} . format ( k table_name namespace ))
def intersect_join_id(self, data_instances):
        LOGGER.info("Join id role is {}".format(self.role))

        sid_hash_pair = None
        if self.use_hash and self.hash_method != "none":
            sid_hash_pair = data_instances.map(
                lambda k, v: (Intersect.hash(k, self.hash_operator, self.salt), k))
            data_sid = sid_hash_pair.mapValues(lambda v: 1)
        else:
            data_sid = data_instances.mapValues(lambda v: 1)

        if self.role == consts.HOST:
            send_ids_federation = self.transfer_variable.send_ids_guest
        elif self.role == consts.GUEST:
            send_ids_federation = self.transfer_variable.send_ids_host
        else:
            raise ValueError("Unknown intersect role, please check the code")

        recv_ids_list = send_ids_federation.get(idx=-1)

        ids_list_size = len(recv_ids_list)
        LOGGER.info("Get ids_list from role-send, ids_list size is {}".format(len(recv_ids_list)))

        if ids_list_size == 1:
            hash_intersect_ids = recv_ids_list[0].join(data_sid, lambda i, d: "intersect_id")
        elif ids_list_size > 1:
            hash_intersect_ids_list = []
            for ids in recv_ids_list:
                hash_intersect_ids_list.append(ids.join(data_sid, lambda i, d: "intersect_id"))
            hash_intersect_ids = self.get_common_intersection(hash_intersect_ids_list)
        else:
            hash_intersect_ids = None
        LOGGER.info("Finish intersect_ids computing")

        if self.sync_intersect_ids:
            if self.role == consts.GUEST:
                intersect_ids_federation = self.transfer_variable.intersect_ids_guest
                send_role = consts.HOST
            elif self.role == consts.HOST:
                intersect_ids_federation = self.transfer_variable.intersect_ids_host
                send_role = consts.GUEST
            else:
                raise ValueError("Unknown intersect role, please check the code")

            intersect_ids_federation.remote(hash_intersect_ids,
                                            role=send_role,
                                            idx=-1)
            LOGGER.info("Remote intersect ids to role-send")

            if self.role == consts.HOST and len(self.host_party_id_list) > 1:
                LOGGER.info(f"raw intersect join role is host,"
                            f"and has {self.host_party_id_list} hosts, get the final intersect_ids from guest")
                hash_intersect_ids = self.transfer_variable.sync_intersect_ids_multi_hosts.get(idx=0)

        if sid_hash_pair:
            hash_intersect_ids_map = hash_intersect_ids.join(sid_hash_pair, lambda r, s: s)
            intersect_ids = hash_intersect_ids_map.map(lambda k, v: (v, 'intersect_id'))
        else:
            intersect_ids = hash_intersect_ids

        """
        if self.task_version_id is not None:
            namespace = "#".join([str(self.guest_party_id), str(self.host_party_id), "mountain"])
            for k, v in enumerate(recv_ids_list):
                table_name = '_'.join([self.task_version_id, str(k)])
                self.tracker.job_tracker.save_as_table(v, table_name, namespace)
                LOGGER.info("save guest_{}'s id in name:{}, namespace:{}".format(k, table_name, namespace))
        """
        return intersect_ids
def intersect_join_id ( self , data_instances ) : LOGGER . info ( "Join id role is {}" . format ( self . role ) ) sid_hash_pair = None if self . use_hash and self . hash_method != "none" : sid_hash_pair = data_instances . map ( lambda k , v : ( Intersect . hash ( k , self . hash_operator , self . salt ) , k ) ) data_sid = sid_hash_pair . mapValues ( lambda v : 1 ) else : data_sid = data_instances . mapValues ( lambda v : 1 ) if self . role == consts . HOST : send_ids_federation = self . transfer_variable . send_ids_guest elif self . role == consts . GUEST : send_ids_federation = self . transfer_variable . send_ids_host else : raise ValueError ( "Unknown intersect role, please check the code" ) recv_ids_list = send_ids_federation . get ( idx = - 1 ) ids_list_size = len ( recv_ids_list ) LOGGER . info ( "Get ids_list from role-send, ids_list size is {}" . format ( len ( recv_ids_list ) ) ) if ids_list_size == 1 : hash_intersect_ids = recv_ids_list [ 0 ] . join ( data_sid , lambda i , d : "intersect_id" ) elif ids_list_size > 1 : hash_intersect_ids_list = [ ] for ids in recv_ids_list : hash_intersect_ids_list . append ( ids . join ( data_sid , lambda i , d : "intersect_id" ) ) hash_intersect_ids = self . get_common_intersection ( hash_intersect_ids_list ) else : hash_intersect_ids = None LOGGER . info ( "Finish intersect_ids computing" ) if self . sync_intersect_ids : if self . role == consts . GUEST : intersect_ids_federation = self . transfer_variable . intersect_ids_guest send_role = consts . HOST elif self . role == consts . HOST : intersect_ids_federation = self . transfer_variable . intersect_ids_host send_role = consts . GUEST else : raise ValueError ( "Unknown intersect role, please check the code" ) intersect_ids_federation . remote ( hash_intersect_ids , role = send_role , idx = - 1 ) LOGGER . info ( "Remote intersect ids to role-send" ) if self . role == consts . HOST and len ( self . host_party_id_list ) > 1 : LOGGER . info ( f"raw intersect join role is host," f"and has {self.host_party_id_list} hosts, get the final intersect_ids from guest" ) hash_intersect_ids = self . transfer_variable . sync_intersect_ids_multi_hosts . get ( idx = 0 ) if sid_hash_pair : hash_intersect_ids_map = hash_intersect_ids . join ( sid_hash_pair , lambda r , s : s ) intersect_ids = hash_intersect_ids_map . map ( lambda k , v : ( v , 'intersect_id' ) ) else : intersect_ids = hash_intersect_ids return intersect_ids