RefineMiniBatch.hpp

Go to the documentation of this file.

#ifndef KMEANS_REFINE_MINIBATCH_HPP
#define KMEANS_REFINE_MINIBATCH_HPP
 
#include <vector>
#include <algorithm>
#include <numeric>
#include <cstdint>
#include <stdexcept>
#include <limits>
#include <random>
#include <type_traits>
 
#include "aarand/aarand.hpp"
 
#include "Refine.hpp"
#include "Details.hpp"
#include "QuickSearch.hpp"
#include "is_edge_case.hpp"
#include "parallelize.hpp"
 
namespace kmeans {
 
struct RefineMiniBatchOptions {
    int max_iterations = 100;
 
    int batch_size = 500;
 
    double max_change_proportion = 0.01;
 
    int convergence_history = 10;
 
    uint64_t seed = 1234567890u;
 
    int num_threads = 1;
};
 
template<typename Index_, typename Data_, typename Cluster_, typename Float_, typename Matrix_ = Matrix<Index_, Data_> >
class RefineMiniBatch : public Refine<Index_, Data_, Cluster_, Float_, Matrix_> {
public:
    RefineMiniBatch(RefineMiniBatchOptions options) : my_options(std::move(options)) {}
 
    RefineMiniBatch() = default;
 
public:
    RefineMiniBatchOptions& get_options() {
        return my_options;
    }
 
private:
    RefineMiniBatchOptions my_options;
 
public:
    Details<Index_> run(const Matrix_& data, Cluster_ ncenters, Float_* centers, Cluster_* clusters) const {
        Index_ nobs = data.num_observations();
        if (internal::is_edge_case(nobs, ncenters)) {
            return internal::process_edge_case(data, ncenters, centers, clusters);
        }
 
        int iter = 0, status = 0;
        std::vector<uint64_t> total_sampled(ncenters); // holds the number of sampled observations across iterations, so we need a large integer.
        std::vector<Cluster_> previous(nobs);
        std::vector<uint64_t> last_changed(ncenters), last_sampled(ncenters); // holds the number of sampled/changed observation for the last few iterations.
 
        Index_ actual_batch_size = nobs;
        typedef typename std::conditional<std::is_signed<Index_>::value, int, unsigned int>::type SafeCompInt; // waiting for C++20's comparison functions...
        if (static_cast<SafeCompInt>(actual_batch_size) > my_options.batch_size) {
            actual_batch_size = my_options.batch_size;
        }
        std::vector<Index_> chosen(actual_batch_size);
        std::mt19937_64 eng(my_options.seed);
 
        size_t ndim = data.num_dimensions();
        internal::QuickSearch<Float_, Cluster_> index;
 
        for (iter = 1; iter <= my_options.max_iterations; ++iter) {
            aarand::sample(nobs, actual_batch_size, chosen.data(), eng);
            if (iter > 1) {
                for (auto o : chosen) {
                    previous[o] = clusters[o];
                }
            }
 
            index.reset(ndim, ncenters, centers);
            parallelize(my_options.num_threads, actual_batch_size, [&](int, Index_ start, Index_ length) -> void {
                auto work = data.new_extractor(chosen.data() + start, length);
                for (Index_ s = start, end = start + length; s < end; ++s) {
                    auto ptr = work->get_observation();
                    clusters[chosen[s]] = index.find(ptr);
                }
            });
 
            // Updating the means for each cluster.
            auto work = data.new_extractor(chosen.data(), actual_batch_size);
            for (auto o : chosen) {
                const auto c = clusters[o];
                auto& n = total_sampled[c];
                ++n;
 
                Float_ mult = static_cast<Float_>(1)/static_cast<Float_>(n);
                auto ccopy = centers + static_cast<size_t>(c) * ndim; // cast to size_t to avoid overflow.
                auto ocopy = work->get_observation();
 
                for (size_t d = 0; d < ndim; ++d) {
                    ccopy[d] += (static_cast<Float_>(ocopy[d]) - ccopy[d]) * mult; // cast to ensure consistent precision regardless of Matrix_::data_type.
                }
            }
 
            // Checking for updates.
            if (iter != 1) {
                for (auto o : chosen) {
                    auto p = previous[o];
                    ++(last_sampled[p]);
                    auto c = clusters[o];
                    if (p != c) {
                        ++(last_sampled[c]);
                        ++(last_changed[p]);
                        ++(last_changed[c]);
                    }
                }
 
                if (iter % my_options.convergence_history == 1) {
                    bool too_many_changes = false;
                    for (Cluster_ c = 0; c < ncenters; ++c) {
                        if (static_cast<double>(last_changed[c]) >= static_cast<double>(last_sampled[c]) * my_options.max_change_proportion) {
                            too_many_changes = true;
                            break;
                        }
                    }
 
                    if (!too_many_changes) {
                        break;
                    }
                    std::fill(last_sampled.begin(), last_sampled.end(), 0);
                    std::fill(last_changed.begin(), last_changed.end(), 0);
                }
            }
        }
 
        if (iter == my_options.max_iterations + 1) {
            status = 2;
        }
 
        // Run through all observations to make sure they have the latest cluster assignments.
        index.reset(ndim, ncenters, centers);
        parallelize(my_options.num_threads, nobs, [&](int, Index_ start, Index_ length) -> void {
            auto work = data.new_extractor(start, length);
            for (Index_ s = start, end = start + length; s < end; ++s) {
                auto ptr = work->get_observation();
                clusters[s] = index.find(ptr);
            }
        });
 
        std::vector<Index_> cluster_sizes(ncenters);
        for (Index_ o = 0; o < nobs; ++o) {
            ++cluster_sizes[clusters[o]];
        }
 
        internal::compute_centroids(data, ncenters, centers, clusters, cluster_sizes);
        return Details<Index_>(std::move(cluster_sizes), iter, status);
    }
};
 
}
 
#endif