<jupyter_start><jupyter_text>Cirrus Demo Simple ExampleThis will run a simple logistic regression on the Criteo Kaggle dataset<jupyter_code>import cirrus
from plotly import __version__
from plotly.offline import download_plotlyjs, init_notebook_mode, plot, iplot
from plotly import tools
import plotly.tools as tls   
import plotly.plotly as py
import plotly.graph_objs as go
import sys
print(sys.version)

from __future__ import print_function
import ipywidgets as widgets
from IPython.display import Image, display, clear_output
from plotly.tools import FigureFactory as FF
from plotly.widgets import GraphWidget
import ipywidgets as widgets

import time
import threading
import random

class mock():
    
    def __init__(self, strid):
        self.pipe = py.Stream(strid)
        self.pipe.open()
        self.kill_sig = threading.Event()
    
    def start_thread(self):
        
        def num_producer():
            start_time = time.time()
            
            while not self.kill_sig.is_set():
                time.sleep(0.5)
                now_time = time.time()
                integer = random.random()
                self.pipe.write(dict(x = now_time - start_time, y = integer))
        
        self.thr = threading.Thread(target=num_producer)
        self.thr.start()
        
    def kill(self):
        print("Mock received kill command")
        self.kill_sig.set()
        self.thr.join()
        self.pipe.close()
        print("Mock is dead")


    
import collab



data_bucket = 'cirrus-criteo-kaggle-19b-random'
model = 'model_v1'

lr_task = collab.CollaborativeFiltering(
             # number of workers
             n_workers = 5,
             # number of parameter servers
             n_ps = 2,
             # worker size in MB
             worker_size = 128,
             # path to s3 bucket with input dataset
             dataset = data_bucket,
             # sgd update LR and epsilon
             learning_rate=0.01,
             epsilon=0.0001,
             progress_callback = print,
             # stop workload after these many seconds
             timeout = 0,
             # stop workload once we reach this loss
             threshold_loss=0,
             # resume execution from model stored in this s3 bucket
             resume_model = model,
             # aws key name
             key_name='mykey',
             # path to aws key
             key_path='/home/camus/Downloads/mykey.pem',
             # ip where ps lives
             ps_ip_public='ec2-54-188-0-164.us-west-2.compute.amazonaws.com',
             ps_ip_private='172.31.26.54',
             # username of VM
             ps_username='ubuntu',
             # choose between adagrad, sgd, nesterov, momentum
             opt_method = 'adagrad',
             # checkpoint model every x secs
             checkpoint_model = 60,
             #
             minibatch_size=20,
             # model size
             model_bits=19,
             # whether to filter gradient weights
             use_grad_threshold=False,
             # threshold value
             grad_threshold=0.001,
             # range of training minibatches
             train_set=(0,824),
             # range of testing minibatches
             test_set=(835,840)
             )

lr_task1 = collab.CollaborativeFiltering(
             # number of workers
             n_workers = 5,
             # number of parameter servers
             n_ps = 2,
             # worker size in MB
             worker_size = 128,
             # path to s3 bucket with input dataset
             dataset = data_bucket,
             # sgd update LR and epsilon
             learning_rate=0.01,
             epsilon=0.0001,
             progress_callback = print,
             # stop workload after these many seconds
             timeout = 0,
             # stop workload once we reach this loss
             threshold_loss=0,
             # resume execution from model stored in this s3 bucket
             resume_model = model,
             # aws key name
             key_name='mykey',
             # path to aws key
             key_path='/home/camus/Downloads/mykey.pem',
             # ip where ps lives
             ps_ip_public='ec2-54-188-0-164.us-west-2.compute.amazonaws.com',
             ps_ip_private='172.31.26.54',
             # username of VM
             ps_username='ubuntu',
             # choose between adagrad, sgd, nesterov, momentum
             opt_method = 'adagrad',
             # checkpoint model every x secs
             checkpoint_model = 60,
             #
             minibatch_size=20,
             # model size
             model_bits=19,
             # whether to filter gradient weights
             use_grad_threshold=False,
             # threshold value
             grad_threshold=0.001,
             # range of training minibatches
             train_set=(0,824),
             # range of testing minibatches
             test_set=(835,840)
             )


#model, loss = lr_task.wait()

stream_ids = tls.get_credentials_file()['stream_ids']

# Get stream id from stream id list 
stream_id0 = stream_ids[0]
stream_id1 = stream_ids[1]
stream_id2 = stream_ids[2]
stream_id3 = stream_ids[3]


stream_0 = go.Stream(
    token=stream_id0,
    maxpoints=80
)

stream_1 = go.Stream(
    token=stream_id1,
    maxpoints=80
)

stream_2 = go.Stream(
    token=stream_id2,
    maxpoints=80
)

stream_3 = go.Stream(
    token=stream_id3,
    maxpoints=80
)

trace0 = go.Scatter(
    x=[],
    y=[],
    mode='lines+markers',
    stream=stream_0         # (!) embed stream id, 1 per trace
)

trace1 = go.Scatter(
    x=[],
    y=[],
    xaxis='x2',
    yaxis='y2',
    stream=stream_1
)
trace2 = go.Scatter(
    x=[],
    y=[],
    xaxis='x3',
    yaxis='y3',
    stream=stream_2
)
trace3 = go.Scatter(
    x=[],
    y=[],
    xaxis='x4',
    yaxis='y4',
    stream=stream_3
)
data = [trace0, trace1, trace2, trace3]

layout = go.Layout(
    xaxis=dict(
        domain=[0, 0.45]
    ),
    yaxis=dict(
        domain=[0, 0.45]
    ),
    xaxis2=dict(
        domain=[0.55, 1]
    ),
    xaxis3=dict(
        domain=[0, 0.45],
        anchor='y3'
    ),
    xaxis4=dict(
        domain=[0.55, 1],
        anchor='y4'
    ),
    yaxis2=dict(
        domain=[0, 0.45],
        anchor='x2'
    ),
    yaxis3=dict(
        domain=[0.55, 1]
    ),
    yaxis4=dict(
        domain=[0.55, 1],
        anchor='x4'
    )
)

    
fig = go.Figure(data=data, layout=layout)
url = py.plot(fig, filename='multiple-subplots', auto_open=False)
print(url)
g = GraphWidget(url)
display(g)
def message_handler(widget, msg):
    print(msg)
    global lr_task
    if msg[0]['curveNumber'] == 0:
        lr_task.kill()
    if msg[0]['curveNumber'] == 1:
        lr_task1.kill()

pipe0 = py.Stream(stream_id0)
pipe0.open()

pipe1 = py.Stream(stream_id1)
pipe1.open()

def progress_callback0(time_loss, cost, task):
    global pipe0
    print("Current training loss:", time_loss, "current cost ($): ", cost)
    pipe0.write(dict(x=time_loss[0], y=time_loss[1]))
    
def progress_callback1(time_loss, cost, task):
    global pipe1
    print("Current training loss:", time_loss, "current cost ($): ", cost)
    pipe1.write(dict(x=time_loss[0], y=time_loss[1]))


lr_task.progress_callback = progress_callback0
lr_task1.progress_callback = progress_callback1


g.on_click(message_handler)


button = widgets.Button(description="Halt!")
display(button)

def on_button_clicked(b):
    lr_task.kill()
    lr_task1.kill()


button.on_click(on_button_clicked)


    

try:
    lr_task.run()
    lr_task1.run()
except KeyboardInterrupt:
    lr_task.kill()
    lr_task1.kill()
    
    

mock_obj.kill()<jupyter_output>Mock received kill command
Mock is dead
StarcoderdataJupyterStructuredCleanDedup