<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 Distributed Hyper-parameter searching
Step1:  Create and launch AWS instances.
Step2:  MNIST dataset
Grayscale hand-written digits
Step3:  Train a NN to predict the numbers (as simple as it gets)
This also demonstrates the time problem of adjusting hyper-parameters
Step4:  Train with some given parameters...
Step5:  Alright, how about something else...
Step6:  and now something different than that..
Step7:  Issue
Step8:  This will take a while, even if using all cores on a local machine; let's distribute the workload
Before executing the next few blocks, make sure the instances are ready to connect. If launched with the non-blocking thread, we can check if it's done with .is_alive()
Step9:  Connect to the resulting scheduler
Step14:  Define functions which will be distributed to workers...
Step15:  Run test functions...
Step16:  Distribute the actual work
Step17:  Also create a distributed queue system...
Step18:  queue is currently empty...
Step19:  Start submitting jobs to the queue with a thread
Step20:  and begin receiving the results...
<ASSISTANT_TASK:>
Python Code:
%matplotlib inline
# Hide info messages from paramiko
import logging
logging.basicConfig()
logger = logging.getLogger()
logger.setLevel(logging.WARN)
import time
import random
import threading
import pandas as pd
import numpy as np
import plotly.plotly as py
import plotly.graph_objs as go
import matplotlib.pyplot as plt
plt.rcParams['figure.figsize'] = (2, 2)
from distributed import progress, Client
from pprint import pprint
from sklearn.decomposition import PCA
from sklearn.neural_network import MLPClassifier
from sklearn.model_selection import cross_val_score
from sklearn.datasets import load_digits
from cclyde.cluster import Cluster
Explanation: Distributed Hyper-parameter searching
End of explanation
cluster = Cluster(key_name='default_windows', n_nodes=16, cluster_name='default', instance_type='t2.medium')
cluster.configure()
cluster.launch_instances_nonblocking()
Explanation: Create and launch AWS instances.
End of explanation
X, y = load_digits(return_X_y=True)
X = np.asarray([x.flatten() for x in X])
for i in range(3):
    plt.imshow(X[i].reshape((8, 8)), cmap='Greys_r')
    plt.title('Digit: {}'.format(y[i]))
    plt.show()
Explanation: MNIST dataset
Grayscale hand-written digits
End of explanation
pca = PCA(n_components=30)
print 'Features before: ', X.shape[1]
X = pca.fit_transform(X)
print 'Features after: ', X.shape[1]
print '{}% Explained Variance'.format(round(sum(pca.explained_variance_ratio_) * 100, 1))
Explanation: Train a NN to predict the numbers (as simple as it gets)
This also demonstrates the time problem of adjusting hyper-parameters
End of explanation
lr = MLPClassifier(hidden_layer_sizes=(10, 5), batch_size=10,
                   solver='sgd', learning_rate_init=0.01, early_stopping=True)
start = time.time()
scores = cross_val_score(estimator=lr,
                         X=X, 
                         y=y,
                         cv=5)
print("\nAccuracy: {}% (+/- {})".format(round(scores.mean() * 100, 2), round(scores.std(), 3) * 2))
print('Finished in {}sec\n'.format(round(time.time() - start, 2)))
Explanation: Train with some given parameters...
End of explanation
lr = MLPClassifier(hidden_layer_sizes=(10, 10,), batch_size=100,
                   solver='sgd', learning_rate_init=0.01, early_stopping=True)
start = time.time()
scores = cross_val_score(estimator=lr,
                         X=X, 
                         y=y,
                         cv=5)
print("\nAccuracy: {}% (+/- {})".format(round(scores.mean() * 100, 2), round(scores.std(), 3) * 2))
print('Finished in {}sec\n'.format(round(time.time() - start, 2)))
Explanation: Alright, how about something else...
End of explanation
lr = MLPClassifier(hidden_layer_sizes=(10, 10, 10,), batch_size=100,
                   solver='sgd', learning_rate_init=0.01, early_stopping=True)
start = time.time()
scores = cross_val_score(estimator=lr,
                         X=X, 
                         y=y,
                         cv=5)
print("\nAccuracy: {}% (+/- {})".format(round(scores.mean() * 100, 2), round(scores.std(), 3) * 2))
print('Finished in {}sec\n'.format(round(time.time() - start, 2)))
Explanation: and now something different than that..
End of explanation
# Define hyper parameter ranges
batch_sizes = np.linspace(start=5, stop=750, num=50, dtype=np.int64)
n_layers = range(1, 8, 1)
# Make a list of all combinations
params = []
for batch_size in batch_sizes:
    for n_layer in n_layers:
        n_neuron = np.random.randint(low=5, high=200)
        params.append({'batch_size': batch_size,
                       'hidden_layer_sizes': tuple(n_neuron for _ in range(n_layer)),
                       'solver': 'sgd',
                       'learning_rate_init': 0.01,
                       'early_stopping': True
                      })
print '{} different combinations.'.format(len(params))
pprint(params[:2])
Explanation: Issue: What hyper params are best?
Train for all/most?
End of explanation
print 'Lauching thread is alive: ', cluster.instance_launching_thread.is_alive()
cluster.install_anaconda()
cluster.install_python_packages(['scikit-learn', 'numpy', 'pandas', 'dask', 'futures'], method='conda')
scheduler_address = cluster.launch_dask()
Explanation: This will take a while, even if using all cores on a local machine; let's distribute the workload
Before executing the next few blocks, make sure the instances are ready to connect. If launched with the non-blocking thread, we can check if it's done with .is_alive()
End of explanation
c = Client(address=scheduler_address)
c
Explanation: Connect to the resulting scheduler
End of explanation
def get_data(kwargs):
    
    Function which gets data and performs PCA on it.
    
    from sklearn.datasets import load_digits
    from sklearn.decomposition import PCA
    import numpy as np
    
    X, y = load_digits(return_X_y=True)
    X = np.asarray([x.flatten() for x in X])
    pca = PCA(n_components=30)
    X = pca.fit_transform(X)
    
    return (kwargs, X, y)
def model_tester(package):
    
    Function which is mapped to cluster. Passes kwargs to model to be trained.
    Returns score based on those kwargs.
    
    
    kwargs, X, y = package
    
    import time
    import numpy as np
    from sklearn.neural_network import MLPClassifier
    from sklearn.model_selection import cross_val_score
    
    # Initialize model with given kwargs
    lr = MLPClassifier(**kwargs)
    scores = cross_val_score(estimator=lr,
                             X=X, 
                             y=y,
                             cv=5)
    return (kwargs, scores.mean(), scores.std())
def score_combiner(package):
    
    Not needed, but more functions == more pretty colors
    
    import time
    import random
    time.sleep(random.random())
    
    kwargs, score_m, score_std = package
    kwargs.update({'score': score_m, 'std': score_std})
    
    return kwargs
def double(n):
    '''
    Useless worker function # 1
    '''
    import time
    import random
    import sklearn
    time.sleep(random.random())
    return n * 2, 2
def add_two(package):
    
    Useless worker function # 2
    
    n, n2 = package
    import time
    import random
    time.sleep(random.random())
    return n + n2
    
Explanation: Define functions which will be distributed to workers...
End of explanation
futures = c.map(double, range(250))
futures = c.map(add_two, futures)
progress(futures)
Explanation: Run test functions...
End of explanation
futures = c.map(get_data, params)
futures = c.map(model_tester, futures)
futures = c.map(score_combiner, futures)
progress(futures)
results = c.gather(futures)
df = pd.DataFrame(results)
df['n_layers'] = df.hidden_layer_sizes.map(lambda _tuple: len(_tuple)) 
df['n_neurons'] = df.hidden_layer_sizes.map(lambda _tuple: _tuple[0])
df.head()
df.n_layers.unique()
data = []
for n_layers in df.n_layers.unique():
    
    temp = df[df.n_layers == n_layers]
    
    trace = go.Scatter(
        x = temp.n_neurons,
        y = temp.n_layers,
        mode='markers',
        text=['{}%<br>Layers: {}'.format(round(v * 100, 2), l) 
              for v, l in zip(temp.score.values, temp.n_layers.values)],
        name='{} layers'.format(n_layers),
        marker=dict(
            size=temp.batch_size / 20.0,
            color = temp.score, #set color equal to a variable
            colorscale='Viridis',
            showscale=False
        )
    )
    data.append(trace)
layout = dict(title = 'Best performing models.<br>(size = batch size)',
              xaxis = dict(zeroline = False, title='Neuron Count'),
              yaxis = dict(zeroline = False, title='Layer Count'),
             )
fig = dict(data=data, layout=layout)
py.iplot(fig, filename='styled-scatter')
df.ix[df.score.argmax(), :]
Explanation: Distribute the actual work
End of explanation
from Queue import Queue
local_q = Queue()
remote_q = c.scatter(local_q)
def long_calc1(n):
    
    import time
    import random
    
    time.sleep(random.random())
    
    return n + 2
def long_calc2(n):
    
    import time
    import random
    
    time.sleep(random.random())
    
    return n * 2
def long_calc3(n):
    
    import time
    import random
    
    time.sleep(random.random())
    
    return n - 2
long_calc1_q = c.map(long_calc1, remote_q)
long_calc2_q = c.map(long_calc2, long_calc1_q)
long_calc3_q = c.map(long_calc3, long_calc2_q)
result_q = c.gather(long_calc3_q)
Explanation: Also create a distributed queue system...
End of explanation
result_q.qsize()
Explanation: queue is currently empty...
End of explanation
def start_jobs():
    jobs = range(500)
    for job in jobs:
        time.sleep(random.random())
        local_q.put(job)
        
    return 
thread = threading.Thread(target=start_jobs)
thread.start()
Explanation: Start submitting jobs to the queue with a thread
End of explanation
def get_jobs():
    while True:
        print result_q.get()
        
    return
finish_thread = threading.Thread(target=get_jobs)
finish_thread.start()
cluster.terminate_cluster()
Explanation: and begin receiving the results...
End of explanation
<END_TASK>