| /* ----------------------------------------------------------------------- *//** |
| * |
| * @file lda.cpp |
| * |
| * @brief Functions for Latent Dirichlet Allocation |
| * |
| * @date Dec 13, 2012 |
| *//* ----------------------------------------------------------------------- */ |
| |
| |
| #include <dbconnector/dbconnector.hpp> |
| #include <math.h> |
| #include <iostream> |
| #include <sstream> |
| #include <algorithm> |
| #include <functional> |
| #include <numeric> |
| #include "lda.hpp" |
| |
| namespace madlib { |
| namespace modules { |
| namespace lda { |
| |
| using namespace dbal::eigen_integration; |
| using madlib::dbconnector::postgres::madlib_construct_array; |
| using madlib::dbconnector::postgres::madlib_construct_md_array; |
| |
| /** |
| * @brief This function samples a new topic for a word in a document based on |
| * the topic counts computed on the rest of the corpus. This is the core |
| * function in the Gibbs sampling inference algorithm for LDA. |
| * @param topic_num The number of topics |
| * @param topic The current topic assignment to a word |
| * @param count_d_z The document topic counts |
| * @param count_w_z The word topic counts |
| * @param count_z The corpus topic counts |
| * @param alpha The Dirichlet parameter for the per-doc topic |
| * multinomial |
| * @param beta The Dirichlet parameter for the per-topic word |
| * multinomial |
| * @return retopic The new topic assignment to the word |
| * @note The topic ranges from 0 to topic_num - 1. |
| * |
| * @note For the sake of performance, this function will not check the validity |
| * of parameters. The caller will ensure that the three pointers all have non-null |
| * values and the lengths are the actual lengths of the arrays. And this |
| * function is local to this file only, so this function cannot be maliciously |
| * called by intruders. |
| **/ |
| static int32_t __lda_gibbs_sample( |
| int32_t topic_num, int32_t topic, const int32_t * count_d_z, const int32_t * count_w_z, |
| const int64_t * count_z, double alpha, double beta) |
| { |
| /* The cumulative probability distribution of the topics */ |
| double * topic_prs = new double[topic_num]; |
| if(!topic_prs) |
| throw std::runtime_error("out of memory"); |
| |
| /* Calculate topic (unnormalised) probabilities */ |
| double total_unpr = 0; |
| for (int32_t i = 0; i < topic_num; i++) { |
| int32_t nwz = count_w_z[i]; |
| int32_t ndz = count_d_z[i]; |
| int64_t nz = count_z[i]; |
| |
| /* Adjust the counts to exclude current word's contribution */ |
| if (i == topic) { |
| nwz--; |
| ndz--; |
| nz--; |
| } |
| |
| /* Compute the probability */ |
| // Note that ndz, nwz, nz are non-negative, and topic_num, alpha, and |
| // beta are positive, so the division by zero will not occure here. |
| double unpr = |
| (ndz + alpha) * (static_cast<double>(nwz) + beta) |
| / (static_cast<double>(nz) + topic_num * beta); |
| total_unpr += unpr; |
| topic_prs[i] = total_unpr; |
| } |
| |
| /* Normalise the probabilities */ |
| // Note that the division by zero will not occure here, so no need to check |
| // whether total_unpr is zero |
| for (int32_t i = 0; i < topic_num; i++) |
| topic_prs[i] /= total_unpr; |
| |
| /* Draw a topic at random */ |
| double r = drand48(); |
| int32_t retopic = 0; |
| while (true) { |
| if (retopic == topic_num - 1 || r < topic_prs[retopic]) |
| break; |
| retopic++; |
| } |
| |
| delete[] topic_prs; |
| return retopic; |
| } |
| |
| /** |
| * @brief Get the min value of an array - for parameter checking |
| * @return The min value |
| * @note The caller will ensure that ah is always non-null. |
| **/ |
| template<class T> static T __min( |
| ArrayHandle<T> ah, size_t start, size_t len){ |
| const T * array = ah.ptr() + start; |
| return *std::min_element(array, array + len); |
| } |
| template<class T> static T __min(ArrayHandle<T> ah){ |
| return __min(ah, 0, ah.size()); |
| } |
| |
| /** |
| * @brief Get the max value of an array - for parameter checking |
| * @return The max value |
| * @note The caller will ensure that ah is always non-null. |
| **/ |
| template<class T> static T __max( |
| ArrayHandle<T> ah, size_t start, size_t len){ |
| const T * array = ah.ptr() + start; |
| return *std::max_element(array, array + len); |
| } |
| template<class T> static T __max(ArrayHandle<T> ah){ |
| return __max(ah, 0, ah.size()); |
| } |
| |
| /** |
| * @brief Get the sum of an array - for parameter checking |
| * @return The sum |
| * @note The caller will ensure that ah is always non-null. |
| **/ |
| static int32_t __sum(ArrayHandle<int32_t> ah){ |
| const int32_t * array = ah.ptr(); |
| size_t size = ah.size(); |
| return std::accumulate(array, array + size, static_cast<int32_t>(0)); |
| } |
| |
| /** |
| * @brief This function learns the topics of words in a document and is the |
| * main step of a Gibbs sampling iteration. The word topic counts and |
| * corpus topic counts are passed to this function in the first call and |
| * then transfered to the rest calls through args.mSysInfo->user_fctx for |
| * efficiency. |
| * @param args[0] The unique words in the documents |
| * @param args[1] The counts of each unique words |
| * @param args[2] The topic counts and topic assignments in the document |
| * @param args[3] The model (word topic counts and corpus topic |
| * counts) |
| * @param args[4] The Dirichlet parameter for per-document topic |
| * multinomial, i.e. alpha |
| * @param args[5] The Dirichlet parameter for per-topic word |
| * multinomial, i.e. beta |
| * @param args[6] The size of vocabulary |
| * @param args[7] The number of topics |
| * @param args[8] The number of iterations (=1:training, >1:prediction) |
| * @return The updated topic counts and topic assignments for |
| * the document |
| **/ |
| AnyType lda_gibbs_sample::run(AnyType & args) |
| { |
| ArrayHandle<int32_t> words = args[0].getAs<ArrayHandle<int32_t> >(); |
| ArrayHandle<int32_t> counts = args[1].getAs<ArrayHandle<int32_t> >(); |
| MutableArrayHandle<int32_t> doc_topic = args[2].getAs<MutableArrayHandle<int32_t> >(); |
| double alpha = args[4].getAs<double>(); |
| double beta = args[5].getAs<double>(); |
| int32_t voc_size = args[6].getAs<int32_t>(); |
| int32_t topic_num = args[7].getAs<int32_t>(); |
| int32_t iter_num = args[8].getAs<int32_t>(); |
| size_t model64_size = |
| static_cast<size_t>(voc_size * (topic_num + 1) + 1) * sizeof(int32_t) / sizeof(int64_t); |
| |
| if(alpha <= 0) |
| throw std::invalid_argument("invalid argument - alpha"); |
| if(beta <= 0) |
| throw std::invalid_argument("invalid argument - beta"); |
| if(voc_size <= 0) |
| throw std::invalid_argument( |
| "invalid argument - voc_size"); |
| if(topic_num <= 0) |
| throw std::invalid_argument( |
| "invalid argument - topic_num"); |
| if(iter_num <= 0) |
| throw std::invalid_argument( |
| "invalid argument - iter_num"); |
| |
| if(words.size() != counts.size()) |
| throw std::invalid_argument( |
| "dimensions mismatch: words.size() != counts.size()"); |
| if(__min(words) < 0 || __max(words) >= voc_size) |
| throw std::invalid_argument( |
| "invalid values in words"); |
| if(__min(counts) <= 0) |
| throw std::invalid_argument( |
| "invalid values in counts"); |
| |
| int32_t word_count = __sum(counts); |
| if(doc_topic.size() != (size_t)(word_count + topic_num)) |
| throw std::invalid_argument( |
| "invalid dimension - doc_topic.size() != word_count + topic_num"); |
| if(__min(doc_topic, 0, topic_num) < 0) |
| throw std::invalid_argument("invalid values in topic_count"); |
| if( |
| __min(doc_topic, topic_num, word_count) < 0 || |
| __max(doc_topic, topic_num, word_count) >= topic_num) |
| throw std::invalid_argument( "invalid values in topic_assignment"); |
| |
| if (!args.getUserFuncContext()) { |
| ArrayHandle<int64_t> model64 = args[3].getAs<ArrayHandle<int64_t> >(); |
| if (model64.size() != model64_size) { |
| std::stringstream ss; |
| ss << "invalid dimension: model64.size() = " << model64.size(); |
| throw std::invalid_argument(ss.str()); |
| } |
| if (__min(model64) < 0) { |
| throw std::invalid_argument("invalid topic counts in model"); |
| } |
| |
| int32_t *context = |
| static_cast<int32_t *>( |
| MemoryContextAllocZero( |
| args.getCacheMemoryContext(), |
| model64.size() * sizeof(int64_t) |
| + topic_num * sizeof(int64_t))); |
| memcpy(context, model64.ptr(), model64.size() * sizeof(int64_t)); |
| int32_t *model = context; |
| |
| int64_t *running_topic_counts = reinterpret_cast<int64_t *>( |
| context + model64_size * sizeof(int64_t) / sizeof(int32_t)); |
| for (int i = 0; i < voc_size; i ++) { |
| for (int j = 0; j < topic_num; j ++) { |
| running_topic_counts[j] += model[i * (topic_num + 1) + j]; |
| } |
| } |
| |
| args.setUserFuncContext(context); |
| } |
| |
| int32_t *context = static_cast<int32_t *>(args.getUserFuncContext()); |
| if (context == NULL) { |
| throw std::runtime_error("args.mSysInfo->user_fctx is null"); |
| } |
| int32_t *model = context; |
| int64_t *running_topic_counts = reinterpret_cast<int64_t *>( |
| context + model64_size * sizeof(int64_t) / sizeof(int32_t)); |
| |
| int32_t unique_word_count = static_cast<int32_t>(words.size()); |
| for(int32_t it = 0; it < iter_num; it++){ |
| int32_t word_index = topic_num; |
| for(int32_t i = 0; i < unique_word_count; i++) { |
| int32_t wordid = words[i]; |
| for(int32_t j = 0; j < counts[i]; j++){ |
| int32_t topic = doc_topic[word_index]; |
| int32_t retopic = __lda_gibbs_sample( |
| topic_num, topic, doc_topic.ptr(), |
| model + wordid * (topic_num + 1), |
| running_topic_counts, alpha, beta); |
| doc_topic[word_index] = retopic; |
| doc_topic[topic]--; |
| doc_topic[retopic]++; |
| |
| if(iter_num == 1) { |
| if (model[wordid * (topic_num + 1) + retopic] <= 2e9) { |
| running_topic_counts[topic] --; |
| running_topic_counts[retopic] ++; |
| model[wordid * (topic_num + 1) + topic]--; |
| model[wordid * (topic_num + 1) + retopic]++; |
| } else { |
| model[wordid * (topic_num + 1) + topic_num] = 1; |
| } |
| } |
| word_index++; |
| } |
| } |
| } |
| |
| return doc_topic; |
| } |
| /** |
| * @brief This function assigns topics to words in a document randomly and |
| * returns the topic counts and topic assignments. |
| * @param args[0] The word count in the documents |
| * @param args[1] The number of topics |
| * @result The topic counts and topic assignments |
| * (length = topic_num + word_count) |
| **/ |
| AnyType lda_random_assign::run(AnyType & args) |
| { |
| int32_t word_count = args[0].getAs<int32_t>(); |
| int32_t topic_num = args[1].getAs<int32_t>(); |
| |
| if(word_count < 1) |
| throw std::invalid_argument( "invalid argument - word_count"); |
| if(topic_num < 1) |
| throw std::invalid_argument( "invalid argument - topic_num"); |
| |
| // FIXME: construct_array functions circumvent the abstraction layer. These |
| // should be replaced with appropriate Allocator:: calls. |
| MutableArrayHandle<int32_t> doc_topic( |
| madlib_construct_array( |
| NULL, topic_num + word_count, INT4OID, sizeof(int32_t), true, 'i')); |
| |
| for(int32_t i = 0; i < word_count; i++){ |
| int32_t topic = static_cast<int32_t>(random() % topic_num); |
| doc_topic[topic] += 1; |
| doc_topic[topic_num + i] = topic; |
| } |
| |
| return doc_topic; |
| } |
| |
| /** |
| * @brief This function is the sfunc for the aggregator computing the topic |
| * counts. It scans the topic assignments in a document and updates the word |
| * topic counts. |
| * @param args[0] The state variable, current topic counts |
| * @param args[1] The unique words in the document |
| * @param args[2] The counts of each unique word in the document |
| * @param args[3] The topic assignments in the document |
| * @param args[4] The size of vocabulary |
| * @param args[5] The number of topics |
| * @return The updated state |
| **/ |
| AnyType lda_count_topic_sfunc::run(AnyType & args) |
| { |
| if(args[4].isNull() || args[5].isNull()) |
| throw std::invalid_argument("null parameter - voc_size and/or \ |
| topic_num is null"); |
| |
| if(args[1].isNull() || args[2].isNull() || args[3].isNull()) |
| return args[0]; |
| |
| int32_t voc_size = args[4].getAs<int32_t>(); |
| int32_t topic_num = args[5].getAs<int32_t>(); |
| if(voc_size <= 0) |
| throw std::invalid_argument( |
| "invalid argument - voc_size"); |
| if(topic_num <= 0) |
| throw std::invalid_argument( |
| "invalid argument - topic_num"); |
| |
| ArrayHandle<int32_t> words = args[1].getAs<ArrayHandle<int32_t> >(); |
| ArrayHandle<int32_t> counts = args[2].getAs<ArrayHandle<int32_t> >(); |
| ArrayHandle<int32_t> topic_assignment = args[3].getAs<ArrayHandle<int32_t> >(); |
| if(words.size() != counts.size()) |
| throw std::invalid_argument( |
| "dimensions mismatch - words.size() != counts.size()"); |
| if(__min(words) < 0 || __max(words) >= voc_size) |
| throw std::invalid_argument( |
| "invalid values in words"); |
| if(__min(counts) <= 0) |
| throw std::invalid_argument( |
| "invalid values in counts"); |
| if(__min(topic_assignment) < 0 || __max(topic_assignment) >= topic_num) |
| throw std::invalid_argument("invalid values in topics"); |
| if((size_t)__sum(counts) != topic_assignment.size()) |
| throw std::invalid_argument( |
| "dimension mismatch - sum(counts) != topic_assignment.size()"); |
| |
| // FIXME: construct_array functions circumvent the abstraction layer. These |
| // should be replaced with appropriate Allocator:: calls. |
| MutableArrayHandle<int64_t> state(NULL); |
| int32_t *model; |
| if(args[0].isNull()) { |
| // to store a voc_size x (topic_num+1) integer matrix in |
| // bigint[] (the +1 is for a flag of ceiling the count), |
| // we need padding if the size is odd. |
| // 1. when voc_size * (topic_num + 1) is (2n+1), gives (n+1) |
| // 2. when voc_size * (topic_num + 1) is (2n), gives (n) |
| int dims[1] = {static_cast<int>( (voc_size * (topic_num + 1) + 1) * sizeof(int32_t) / sizeof(int64_t) )}; |
| int lbs[1] = {1}; |
| state = madlib_construct_md_array( |
| NULL, NULL, 1, dims, lbs, INT8OID, sizeof(int64_t), true, 'd'); |
| // the reason we use bigint[] because integer[] has limit on number of |
| // elements and thus cannot be larger than 500MB |
| model = reinterpret_cast<int32_t *>(state.ptr()); |
| } else { |
| state = args[0].getAs<MutableArrayHandle<int64_t> >(); |
| model = reinterpret_cast<int32_t *>(state.ptr()); |
| } |
| |
| int32_t unique_word_count = static_cast<int32_t>(words.size()); |
| int32_t word_index = 0; |
| for(int32_t i = 0; i < unique_word_count; i++){ |
| int32_t wordid = words[i]; |
| for(int32_t j = 0; j < counts[i]; j++){ |
| int32_t topic = topic_assignment[word_index]; |
| if (model[wordid * (topic_num + 1) + topic] <= 2e9) { |
| model[wordid * (topic_num + 1) + topic]++; |
| } else { |
| model[wordid * (topic_num + 1) + topic_num] = 1; |
| } |
| word_index++; |
| } |
| } |
| return state; |
| } |
| |
| /** |
| * @brief This function is the prefunc for the aggregator computing the |
| * topic counts. |
| * @param args[0] The state variable, local topic counts |
| * @param args[1] The state variable, local topic counts |
| * @return The merged state, element-wise sum of two local states |
| **/ |
| AnyType lda_count_topic_prefunc::run(AnyType & args) |
| { |
| MutableArrayHandle<int64_t> state1 = args[0].getAs<MutableArrayHandle<int64_t> >(); |
| ArrayHandle<int64_t> state2 = args[1].getAs<ArrayHandle<int64_t> >(); |
| |
| if(state1.size() != state2.size()) |
| throw std::invalid_argument("invalid dimension"); |
| |
| int32_t *model1 = reinterpret_cast<int32_t *>(state1.ptr()); |
| const int32_t *model2 = reinterpret_cast<const int32_t *>(state2.ptr()); |
| for(size_t i = 0; i < state1.size() * sizeof(int64_t) / sizeof(int32_t); i++) { |
| model1[i] += model2[i]; |
| } |
| |
| return state1; |
| } |
| |
| /** |
| * @brief This function transposes a matrix represented by a 2-D array |
| * @param args[0] The input matrix |
| * return The transposed matrix |
| **/ |
| AnyType lda_transpose::run(AnyType & args) |
| { |
| ArrayHandle<int64_t> matrix = args[0].getAs<ArrayHandle<int64_t> >(); |
| if(matrix.dims() != 2) |
| throw std::domain_error("invalid dimension"); |
| |
| int32_t row_num = static_cast<int32_t>(matrix.sizeOfDim(0)); |
| int32_t col_num = static_cast<int32_t>(matrix.sizeOfDim(1)); |
| |
| // FIXME: construct_array functions circumvent the abstraction layer. These |
| // should be replaced with appropriate Allocator:: calls. |
| int dims[2] = {col_num, row_num}; |
| int lbs[2] = {1, 1}; |
| MutableArrayHandle<int64_t> transposed( |
| madlib_construct_md_array( |
| NULL, NULL, 2, dims, lbs, INT8OID, sizeof(int64_t), true, 'd')); |
| |
| for(int32_t i = 0; i < row_num; i++){ |
| int32_t index = i * col_num; |
| for(int32_t j = 0; j < col_num; j++){ |
| transposed[j * row_num + i] = matrix[index]; |
| index++; |
| } |
| } |
| |
| return transposed; |
| } |
| |
| /** |
| * @brief This structure defines the states used by the following SRF. |
| **/ |
| typedef struct __sr_ctx{ |
| const int32_t * inarray; |
| int32_t maxcall; |
| int32_t dim; |
| int32_t curcall; |
| } sr_ctx; |
| |
| void * lda_unnest_transpose::SRF_init(AnyType &args) |
| { |
| ArrayHandle<int64_t> inarray64 = args[0].getAs<ArrayHandle<int64_t> >(); |
| |
| sr_ctx * ctx = new sr_ctx; |
| ctx->inarray = reinterpret_cast<const int32_t *>(inarray64.ptr()); |
| ctx->maxcall = args[2].getAs<int32_t>(); |
| ctx->dim = args[1].getAs<int32_t>(); |
| ctx->curcall = 0; |
| |
| return ctx; |
| } |
| |
| /** |
| * @brief The function is used to return the next row by the SRF.. |
| **/ |
| AnyType lda_unnest_transpose::SRF_next(void *user_fctx, bool *is_last_call) |
| { |
| sr_ctx * ctx = (sr_ctx *) user_fctx; |
| if (ctx->maxcall == ctx->curcall) { |
| *is_last_call = true; |
| return Null(); |
| } |
| |
| // FIXME: construct_array functions circumvent the abstraction layer. These |
| // should be replaced with appropriate Allocator:: calls. |
| MutableArrayHandle<int32_t> outarray( |
| madlib_construct_array( |
| NULL, ctx->dim, INT4OID, sizeof(int32_t), true, 'i')); |
| |
| for (int i = 0; i < ctx->dim; i ++) { |
| outarray[i] = ctx->inarray[(ctx->maxcall + 1) * i + ctx->curcall]; |
| } |
| |
| ctx->curcall++; |
| *is_last_call = false; |
| |
| return outarray; |
| } |
| // ----------------------------------------------------------------------- |
| |
| void * lda_unnest::SRF_init(AnyType &args) |
| { |
| ArrayHandle<int64_t> inarray64 = args[0].getAs<ArrayHandle<int64_t> >(); |
| |
| sr_ctx * ctx = new sr_ctx; |
| ctx->inarray = reinterpret_cast<const int32_t *>(inarray64.ptr()); |
| ctx->maxcall = args[1].getAs<int32_t>(); |
| ctx->dim = args[2].getAs<int32_t>(); |
| ctx->curcall = 0; |
| |
| return ctx; |
| } |
| |
| /** |
| * @brief The function is used to return the next row by the SRF.. |
| **/ |
| AnyType lda_unnest::SRF_next(void *user_fctx, bool *is_last_call) |
| { |
| sr_ctx * ctx = (sr_ctx *) user_fctx; |
| if (ctx->maxcall == ctx->curcall) { |
| *is_last_call = true; |
| return Null(); |
| } |
| |
| // FIXME: construct_array functions circumvent the abstraction layer. These |
| // should be replaced with appropriate Allocator:: calls. |
| MutableArrayHandle<int32_t> outarray( |
| madlib_construct_array( |
| NULL, ctx->dim, INT4OID, sizeof(int32_t), true, 'i')); |
| |
| for (int i = 0; i < ctx->dim; i ++) { |
| outarray[i] = ctx->inarray[ctx->curcall * (ctx->dim + 1) + i]; |
| } |
| |
| ctx->curcall++; |
| *is_last_call = false; |
| |
| return outarray; |
| } |
| |
| /** |
| * @brief This function is the sfunc of an aggregator computing the |
| * perplexity. |
| * @param args[0] The current state |
| * @param args[1] The unique words in the documents |
| * @param args[2] The counts of each unique words |
| * @param args[3] The topic counts in the document |
| * @param args[4] The model (word topic counts and corpus topic |
| * counts) |
| * @param args[5] The Dirichlet parameter for per-document topic |
| * multinomial, i.e. alpha |
| * @param args[6] The Dirichlet parameter for per-topic word |
| * multinomial, i.e. beta |
| * @param args[7] The size of vocabulary |
| * @param args[8] The number of topics |
| * @return The updated state |
| **/ |
| AnyType lda_perplexity_sfunc::run(AnyType & args){ |
| ArrayHandle<int32_t> words = args[1].getAs<ArrayHandle<int32_t> >(); |
| ArrayHandle<int32_t> counts = args[2].getAs<ArrayHandle<int32_t> >(); |
| ArrayHandle<int32_t> doc_topic_counts = args[3].getAs<ArrayHandle<int32_t> >(); |
| double alpha = args[5].getAs<double>(); |
| double beta = args[6].getAs<double>(); |
| int32_t voc_size = args[7].getAs<int32_t>(); |
| int32_t topic_num = args[8].getAs<int32_t>(); |
| size_t model64_size = static_cast<size_t>(voc_size * (topic_num + 1) + 1) * sizeof(int32_t) / sizeof(int64_t); |
| |
| if(alpha <= 0) |
| throw std::invalid_argument("invalid argument - alpha"); |
| if(beta <= 0) |
| throw std::invalid_argument("invalid argument - beta"); |
| if(voc_size <= 0) |
| throw std::invalid_argument( |
| "invalid argument - voc_size"); |
| if(topic_num <= 0) |
| throw std::invalid_argument( |
| "invalid argument - topic_num"); |
| |
| if(words.size() != counts.size()) |
| throw std::invalid_argument( |
| "dimensions mismatch: words.size() != counts.size()"); |
| if(__min(words) < 0 || __max(words) >= voc_size) |
| throw std::invalid_argument( |
| "invalid values in words"); |
| if(__min(counts) <= 0) |
| throw std::invalid_argument( |
| "invalid values in counts"); |
| |
| if(doc_topic_counts.size() != (size_t)(topic_num)) |
| throw std::invalid_argument( |
| "invalid dimension - doc_topic_counts.size() != topic_num"); |
| if(__min(doc_topic_counts, 0, topic_num) < 0) |
| throw std::invalid_argument("invalid values in doc_topic_counts"); |
| |
| MutableArrayHandle<int64_t> state(NULL); |
| if (args[0].isNull()) { |
| ArrayHandle<int64_t> model64 = args[4].getAs<ArrayHandle<int64_t> >(); |
| |
| if (model64.size() != model64_size) { |
| std::stringstream ss; |
| ss << "invalid dimension: model64.size() = " << model64.size(); |
| throw std::invalid_argument(ss.str()); |
| } |
| if(__min(model64) < 0) { |
| throw std::invalid_argument("invalid topic counts in model"); |
| } |
| |
| // FIXME: construct_array functions circumvent the abstraction layer. These |
| // should be replaced with appropriate Allocator:: calls. |
| state = madlib_construct_array(NULL, |
| static_cast<int>(model64.size()) + |
| topic_num + |
| static_cast<int>(sizeof(double) / sizeof(int64_t)), |
| INT8OID, sizeof(int64_t), true, 'd'); |
| |
| memcpy(state.ptr(), model64.ptr(), model64.size() * sizeof(int64_t)); |
| int32_t *_model = reinterpret_cast<int32_t *>(state.ptr()); |
| int64_t *_total_topic_counts = reinterpret_cast<int64_t *>(state.ptr() + model64.size()); |
| for (int i = 0; i < voc_size; i ++) { |
| for (int j = 0; j < topic_num; j ++) { |
| _total_topic_counts[j] += _model[i * (topic_num + 1) + j]; |
| } |
| } |
| } else { |
| state = args[0].getAs<MutableArrayHandle<int64_t> >(); |
| } |
| |
| int32_t *model = reinterpret_cast<int32_t *>(state.ptr()); |
| int64_t *total_topic_counts = reinterpret_cast<int64_t *>(state.ptr() + model64_size); |
| double *perp = reinterpret_cast<double *>(state.ptr() + state.size() - 1); |
| |
| int32_t n_d = 0; |
| for(size_t i = 0; i < words.size(); i++){ |
| n_d += counts[i]; |
| } |
| |
| for(size_t i = 0; i < words.size(); i++){ |
| int32_t w = words[i]; |
| int32_t n_dw = counts[i]; |
| |
| double sum_p = 0.0; |
| for(int32_t z = 0; z < topic_num; z++){ |
| int32_t n_dz = doc_topic_counts[z]; |
| int32_t n_wz = model[w * (topic_num + 1) + z]; |
| int64_t n_z = total_topic_counts[z]; |
| sum_p += (static_cast<double>(n_wz) + beta) * (n_dz + alpha) |
| / (static_cast<double>(n_z) + voc_size * beta); |
| } |
| sum_p /= (n_d + topic_num * alpha); |
| |
| *perp += n_dw * log(sum_p); |
| } |
| |
| return state; |
| } |
| |
| /** |
| * @brief This function is the prefunc of an aggregator computing the |
| * perplexity. |
| * @param args[0] The local state |
| * @param args[1] The local state |
| * @return The merged state |
| **/ |
| AnyType lda_perplexity_prefunc::run(AnyType & args){ |
| MutableArrayHandle<int64_t> state1 = args[0].getAs<MutableArrayHandle<int64_t> >(); |
| ArrayHandle<int64_t> state2 = args[1].getAs<ArrayHandle<int64_t> >(); |
| double * perp1 = reinterpret_cast<double *>(state1.ptr() + state1.size() - 1); |
| const double * perp2 = reinterpret_cast<const double *>(state2.ptr() + state2.size() - 1); |
| *perp1 += *perp2; |
| return state1; |
| } |
| |
| /** |
| * @brief This function is the finalfunc of an aggregator computing the |
| * perplexity. |
| * @param args[0] The global state |
| * @return The perplexity |
| **/ |
| AnyType lda_perplexity_ffunc::run(AnyType & args){ |
| ArrayHandle<int64_t> state = args[0].getAs<ArrayHandle<int64_t> >(); |
| const double * perp = reinterpret_cast<const double *>(state.ptr() + state.size() - 1); |
| return *perp; |
| } |
| |
| AnyType |
| lda_check_count_ceiling::run(AnyType &args) { |
| ArrayHandle<int64_t> model64 = args[0].getAs<ArrayHandle<int64_t> >(); |
| int voc_size = args[1].getAs<int32_t>(); |
| int topic_num = args[2].getAs<int32_t>(); |
| |
| int example_words_hit_ceiling[10]; |
| int count = 0; |
| const int32_t *model = reinterpret_cast<const int32_t *>(model64.ptr()); |
| for (int wordid = 0; wordid < voc_size; wordid ++) { |
| int flag = model[wordid * (topic_num + 1) + topic_num]; |
| if (flag != 0) { |
| example_words_hit_ceiling[count ++] = wordid; |
| } |
| } |
| |
| if (count == 0) { return Null(); } |
| |
| MutableNativeIntegerVector ret(this->allocateArray<int>(count)); |
| memcpy(ret.data(), example_words_hit_ceiling, count * sizeof(int)); |
| return ret; |
| } |
| |
| AnyType l1_norm_with_smoothing::run(AnyType & args){ |
| MutableArrayHandle<double> arr = args[0].getAs<MutableArrayHandle<double> >(); |
| double smooth = args[1].getAs<double>(); |
| smooth = fabs(smooth); |
| |
| double sum = 0.0; |
| for(size_t i = 0; i < arr.size(); i++) |
| sum += fabs(arr[i]); |
| sum += smooth * static_cast<double>(arr.size()); |
| |
| double inverse_sum = 0.0; |
| if (sum != 0.0) |
| inverse_sum = 1.0 / sum; |
| for(size_t i = 0; i < arr.size(); i++) |
| arr[i] = (arr[i] + smooth) * inverse_sum; |
| return arr; |
| } |
| |
| AnyType lda_parse_model::run(AnyType & args){ |
| ArrayHandle<int64_t> state = args[0].getAs<ArrayHandle<int64_t> >(); |
| int32_t voc_size = args[1].getAs<int32_t>(); |
| int32_t topic_num = args[2].getAs<int32_t>(); |
| |
| const int32_t *model = reinterpret_cast<const int32_t *>(state.ptr()); |
| |
| // FIXME: construct_array functions circumvent the abstraction layer. These |
| // should be replaced with appropriate Allocator:: calls. |
| int dims[2] = {voc_size/2, topic_num}; |
| int lbs[2] = {1, 1}; |
| MutableArrayHandle<int32_t> model_part1( |
| madlib_construct_md_array( |
| NULL, NULL, 2, dims, lbs, INT4OID, sizeof(int32_t), true, 'i')); |
| |
| for(int32_t i = 0; i < voc_size/2; i++){ |
| for(int32_t j = 0; j < topic_num; j++){ |
| model_part1[i * topic_num + j] = model[i * (topic_num+1) + j]; |
| } |
| } |
| |
| // FIXME: construct_array functions circumvent the abstraction layer. These |
| // should be replaced with appropriate Allocator:: calls. |
| int dims2[2] = {voc_size - voc_size/2, topic_num}; |
| MutableArrayHandle<int32_t> model_part2( |
| madlib_construct_md_array( |
| NULL, NULL, 2, dims2, lbs, INT4OID, sizeof(int32_t), true, 'i')); |
| |
| for(int32_t i = voc_size/2; i < voc_size; i++){ |
| for(int32_t j = 0; j < topic_num; j++){ |
| model_part2[(i-voc_size/2) * topic_num + j] = model[i * (topic_num+1) + j]; |
| } |
| } |
| |
| //int dims3[1] = {topic_num}; |
| //int lbs3[1] = {1}; |
| |
| MutableNativeColumnVector total_topic_counts(allocateArray<double>(topic_num)); |
| |
| for (int i = 0; i < voc_size; i ++) { |
| for (int j = 0; j < topic_num; j ++) { |
| total_topic_counts[j] += static_cast<double>(model[i * (topic_num + 1) + j]); |
| } |
| } |
| |
| AnyType tuple; |
| tuple << model_part1 |
| << model_part2 |
| << total_topic_counts; |
| |
| return tuple; |
| } |
| |
| } |
| } |
| } |
