The merged model is not used after upload to redis

[ullash21]

Model run with 2 threads for map and 2 threads for reduce

modified_PLStream.py:371                                     [] - self flag true update model 
modified_PLStream.py:373                                     [] - inside update_model self is: <__main__.unsupervised_OSA object at 0x7f5204399990> 
modified_PLStream.py:317                                     [] - in update_model map
modified_PLStream.py:318                                     [] - None 
modified_PLStream.py:371                                     [] - self flag true update model 
modified_PLStream.py:373                                     [] - inside update_model self is: <__main__.unsupervised_OSA object at 0x7f65a2b9db10>

This shows the two initial unsupervised_OSA objects inside the map function. The self flag is true once for each thread at the beginning

modified_PLStream.py:104                                     [] - Object created and object is: <__main__.unsupervised_OSA object at 0x7f65a2acc8d0>
modified_PLStream.py:188                                     [] - model_merge
modified_PLStream.py:189                                     [] - model merge
modified_PLStream.py:197                                     [] - model_merge model
modified_PLStream.py:104                                     [] - Object created and object is: <__main__.unsupervised_OSA object at 0x7f5204259b90>
modified_PLStream.py:188                                     [] - model_merge
modified_PLStream.py:189                                     [] - model merge
modified_PLStream.py:104                                     [] - Object created and object is: <__main__.unsupervised_OSA object at 0x7f52042c5f10>
modified_PLStream.py:188                                     [] - model_merge
modified_PLStream.py:189                                     [] - model merge
modified_PLStream.py:104                                     [] - Object created and object is: <__main__.unsupervised_OSA object at 0x7f65a2acc310>
modified_PLStream.py:188                                     [] - model_merge
modified_PLStream.py:189                                     [] - model merge
modified_PLStream.py:197                                     [] - model_merge model
modified_PLStream.py:104                                     [] - Object created and object is: <__main__.unsupervised_OSA object at 0x7f65a2abc810>
modified_PLStream.py:188                                     [] - model_merge
modified_PLStream.py:189                                     [] - model merge
modified_PLStream.py:197                                     [] - model_merge model
modified_PLStream.py:104                                     [] - Object created and object is: <__main__.unsupervised_OSA object at 0x7f52042ce950>
modified_PLStream.py:188                                     [] - model_merge
modified_PLStream.py:189                                     [] - model merge
modified_PLStream.py:104                                     [] - Object created and object is: <__main__.unsupervised_OSA object at 0x7f520425a5d0>
modified_PLStream.py:188                                     [] - model_merge
modified_PLStream.py:189                                     [] - model merge
modified_PLStream.py:104                                     [] - Object created and object is: <__main__.unsupervised_OSA object at 0x7f52042c2f10>
modified_PLStream.py:188                                     [] - model_merge
modified_PLStream.py:189                                     [] - model merge

The above shows object created during model merge with different hashes

         # code in __init__(self)
        self.flag = True

        # code in upload_model()
        if self.flag: 
            logging.warning("self flag true update model")
            self.model_to_train = self.load_model()
            logging.warning("inside update_model self is: " + str(self))
            self.flag = False
           ........

        #code in model_merge() 
        self.save_model(model_new)
        self.flag = True

In all the instances, the self flag are set locally. These are set and altered only in initialisation of the object, in update_model function and merge_model function.

@ShuhaoZhangTony @HuilinWu2

[ullash21]

logfile.txt

[ullash21]

PLStream code that shows when reduce is called it initialises a new object

ds = ds.map(unsupervised_OSA()).set_parallelism(map_parallelism)
    ds = ds.filter(lambda x: x[0] != 'collecting')
    ds = ds.key_by(lambda x: x[0], key_type=Types.STRING())
    ds = ds.reduce(lambda x, y: (x[0], unsupervised_OSA().model_merge(x, y))).set_parallelism(reduce_parallelism)