I've implemented a self-organising map in Tensorflow's low-level API. After testing with sklearn's Iris data, the results seem correct. I did implement the algorithm also using NumPy before converting it to tf, because I'm new to Tensorflow, and had no idea whether it would work or not. So of course I tried out which would perform better. It turns out the Numpy implementation is far better for a smallish-scale problem.
To me this sounds a bit off. When loading a data set of ~3500x10 to GPU memory and carrying out basically the exact same algorithm with all-tf operations, I was at least expecting some sort of improvement with matrix-matrix broadcasted sums, multiplications and powers.
As I said, I have little experience with Tensorflow, so I started wondering if something is wrong with the code. The result seems good enough, but I'm not aware of many best practices or performance caveats. I'm especially after performance and TF-related feedback, but of course I'll happily take any criticism on my code!
import numpy as np
import tensorflow as tf
from tqdm import tqdm
def learning_rate(epoch: tf.placeholder, max_epochs: int):
with tf.name_scope('learning_rate'):
return tf.exp(-4 * epoch / max_epochs)
def neighbourhood(r: tf.placeholder, epoch: tf.placeholder, max_epochs: int, size: int):
with tf.name_scope('neighbourhood'):
return tf.exp(
- (2 * r / size) ** 2
* (max_epochs / (max_epochs - epoch)) ** 3
)
class SelfOrganisingMap:
def __init__(self, shape: tuple, features: int, *_,
max_epochs: int = None, init: str = 'uniform', learning_rate: float = 0.1):
"""
Self-organising map using TensorFlow.
:param shape: map dimensions
:param features: number of input features
:param _: used to force calling with keyword arguments below
:param max_epochs: used to scale the neighbourhood and learning rate functions
:param init: method of weight initialisation. 'uniform' for drawing from an uniform
distribution between 0..1, 'normal' for drawing from X~N(0,1)
:param learning_rate: initial learning rate multiplier
"""
self._weights = None
self._shape = shape
self._features = features
self._neighbour_shape = (len(shape),) + tuple(1 for _ in shape) + (-1,)
self._epochs = 0
self._max_epochs = max_epochs
self._initial_lr = learning_rate
if init == 'uniform':
self._initialiser = tf.random_uniform_initializer
elif init == 'normal':
self._initialiser = tf.random_normal_initializer
else:
raise AssertionError('Unknown weights initialiser type "%s"!' % init)
@property
def weights(self):
if self._weights is None:
raise ValueError('Map not fitted!')
return self._weights
@property
def initialiser(self):
if self._weights is None:
return self._initialiser
else:
return tf.convert_to_tensor(self._weights)
@property
def shape(self): return self._shape
@property
def n_nodes(self): return int(np.prod(self.shape))
@property
def features(self): return self._features
@property
def epochs(self): return self._epochs
@property
def max_epochs(self): return self._max_epochs
def project(self, data: np.ndarray) -> np.array:
"""
Project data onto the map. NumPy implementation for simplicity.
:param data: samples
:return: node indices
"""
diff = self.weights - data
dist = np.sum(diff ** 2, axis=-1, keepdims=True)
return np.array(np.unravel_index(
np.argmin(dist.reshape((-1, data.shape[0])), axis=0), self.shape
))
def train(self, x: np.ndarray, epochs: int, batch_size: int = 1) -> None:
"""
Create training graph and train SOM.
:param x: training data
:param epochs: number of epochs to train
:param batch_size: number of training examples per step
:return: None
"""
graph = tf.Graph()
sess = tf.Session(graph=graph)
x = x.astype(np.float64)
if x.shape[0] % batch_size != 0:
raise ValueError('Bad batch_size, last batch would be incomplete!')
# Construct graph
with graph.as_default():
indices = tf.convert_to_tensor(np.expand_dims(
np.indices(self.shape, dtype=np.float64), axis=-1
))
weights = tf.get_variable(
'weights', (*self.shape, 1, self.features), initializer=self.initialiser, dtype=tf.float64
)
with tf.name_scope('data'):
data = tf.data.Dataset.from_tensor_slices(x)
data = data.shuffle(buffer_size=10000).repeat(epochs)
data = data.batch(batch_size, drop_remainder=True)
data = data.make_one_shot_iterator().get_next()
with tf.name_scope('winner'):
diff = weights - data
dist = tf.reduce_sum(diff ** 2, axis=-1, keepdims=True)
w_ix = tf.argmin(tf.reshape(dist, (self.n_nodes, data.shape[0])), axis=0)
winner_op = tf.convert_to_tensor(tf.unravel_index(w_ix, self.shape))
with tf.name_scope('update'):
curr_epoch = tf.placeholder(dtype=tf.int64, shape=())
idx_diff = indices - tf.reshape(tf.cast(
winner_op, dtype=tf.float64
), shape=self._neighbour_shape)
idx_dist = tf.norm(idx_diff, axis=0)
l_rate = learning_rate(curr_epoch, self.max_epochs)
n_hood = neighbourhood(
idx_dist, curr_epoch, self.max_epochs, max(self.shape)
)
update = diff * l_rate * tf.expand_dims(n_hood, axis=-1)
update_op = weights.assign(
weights - self._initial_lr * tf.reduce_sum(update, axis=-2, keepdims=True)
)
init = tf.global_variables_initializer()
# Initialise all variables
sess.run(init)
batches = int(np.ceil(x.shape[0] // batch_size))
for i in tqdm(range(epochs)):
for b in range(batches):
sess.run(update_op, feed_dict={
curr_epoch: self.epochs + i
})
self._weights = sess.run(weights)
self._epochs += epochs
Here's a short test snippet as well:
from sklearn.utils import shuffle
from sklearn.datasets import load_iris
from sklearn.preprocessing import RobustScaler
a, _ = load_iris(True)
a, _ = shuffle(a, _)
a = RobustScaler().fit_transform(a)
epochs = 100
som = SelfOrganisingMap((100, 100), 4, max_epochs=epochs, init='normal')
som.train(a, epochs, batch_size=1)
