LostTech.TensorFlow : API Documentation

A model for Weighted Alternating Least Squares matrix factorization.

It minimizes the following loss function over U, V: $$ \|\sqrt W \odot (A - U V^T)\|_F^2 + \lambda (\|U\|_F^2 + \|V\|_F^2) $$ where, A: input matrix, W: weight matrix. Note that the (element-wise) square root of the weights is used in the objective function. U, V: row_factors and column_factors matrices, \\(\lambda)\\: regularization. Also we assume that W is of the following special form: \\( W_{ij} = W_0 + R_i * C_j \\) if \\(A_{ij} \ne 0\\), \\(W_{ij} = W_0\\) otherwise. where, \\(W_0\\): unobserved_weight, \\(R_i\\): row_weights, \\(C_j\\): col_weights.

Note that the current implementation supports two operation modes: The default mode is for the condition where row_factors and col_factors can individually fit into the memory of each worker and these will be cached. When this condition can't be met, setting use_factors_weights_cache to False allows the larger problem sizes with slight performance penalty as this will avoid creating the worker caches and instead the relevant weight and factor values are looked up from parameter servers at each step.

Loss computation: The loss can be computed efficiently by decomposing it into a sparse term and a Gramian term, see wals.md. The loss is returned by the update_{col, row}_factors(sp_input), and is normalized as follows: _, _, unregularized_loss, regularization, sum_weights = update_row_factors(sp_input) if sp_input contains the rows \\({A_i, i \in I}\\), and the input matrix A has n total rows, then the minibatch loss = unregularized_loss + regularization is $$ (\|\sqrt W_I \odot (A_I - U_I V^T)\|_F^2 + \lambda \|U_I\|_F^2) * n / |I| + \lambda \|V\|_F^2 $$ The sum_weights tensor contains the normalized sum of weights \\(sum(W_I) * n / |I|\\).

A typical usage example (pseudocode):

with tf.Graph().as_default(): # Set up the model object. model = tf.contrib.factorization.WALSModel(....)

# To be run only once as part of session initialization. In distributed # training setting, this should only be run by the chief trainer and all # other trainers should block until this is done. model_init_op = model.initialize_op

# To be run once per worker after session is available, prior to # the prep_gramian_op for row(column) can be run. worker_init_op = model.worker_init

# To be run once per iteration sweep before the row(column) update # initialize ops can be run. Note that in the distributed training # situations, this should only be run by the chief trainer. All other # trainers need to block until this is done. row_update_prep_gramian_op = model.row_update_prep_gramian_op col_update_prep_gramian_op = model.col_update_prep_gramian_op

# To be run once per worker per iteration sweep. Must be run before # any actual update ops can be run. init_row_update_op = model.initialize_row_update_op init_col_update_op = model.initialize_col_update_op

# Ops to update row(column). This can either take the entire sparse # tensor or slices of sparse tensor. For distributed trainer, each # trainer handles just part of the matrix. _, row_update_op, unreg_row_loss, row_reg, _ = model.update_row_factors( sp_input=matrix_slices_from_queue_for_worker_shard) row_loss = unreg_row_loss + row_reg _, col_update_op, unreg_col_loss, col_reg, _ = model.update_col_factors( sp_input=transposed_matrix_slices_from_queue_for_worker_shard, transpose_input=True) col_loss = unreg_col_loss + col_reg

...

# model_init_op is passed to Supervisor. Chief trainer runs it. Other # trainers wait. sv = tf.compat.v1.train.Supervisor(is_chief=is_chief, ..., init_op=tf.group(..., model_init_op,...),...) ...

with sv.managed_session(...) as sess: # All workers/trainers run it after session becomes available. worker_init_op.run(session=sess)

...

while i in iterations:

# All trainers need to sync up here. while not_all_ready: wait

# Row update sweep. if is_chief: row_update_prep_gramian_op.run(session=sess) else: wait_for_chief

# All workers run upate initialization. init_row_update_op.run(session=sess)

# Go through the matrix. reset_matrix_slices_queue_for_worker_shard while_matrix_slices: row_update_op.run(session=sess)

# All trainers need to sync up here. while not_all_ready: wait

# Column update sweep. if is_chief: col_update_prep_gramian_op.run(session=sess) else: wait_for_chief

# All workers run upate initialization. init_col_update_op.run(session=sess)

# Go through the matrix. reset_transposed_matrix_slices_queue_for_worker_shard while_transposed_matrix_slices: col_update_op.run(session=sess)