Abstract:
Latent Dirichlet allocation (LDA) analysis on a dataset is performed on an MPP relational database by distributing subsets of said dataset to a plurality of segments of the MPP database, and performing LDA analysis in parallel on the respective subsets on the plurality of segments using Gibbs sampling. An object library on each segment provides executable objects of user defined functions that can be called by an SQL query when the query requires functionality provided by an object.

Description:
BACKGROUND 
     This invention relates generally to latent Dirichlet allocation (“LDA”) analysis of a dataset to discover themes or topics, and more particularly to parallel LDA analysis of a distributed dataset comprising a large collection of unstructured data, referred to herein as documents, in a shared-nothing massively parallel processing (MPP) database. 
     Documents of a dataset can be represented as random mixtures of latent topics, where each topic may be characterized by a probability distribution over a vocabulary of data elements such as words. Documents comprise collections of words, and each document may comprise multiple topics. Given a large corpus of text, i.e., a dataset, LDA can infer a set of latent topics from the corpus, each topic being represented as a multinomial distribution over words denoted as P(w/z), and can infer the topic distribution for each document represented as a multinomial distribution over topics denoted as P(z/d). All of the documents in a corpus share the same set of topics, but each document has a different mix (distribution) of topics. Gibbs sampling has been widely used for the inference of LDA because it is simple, fast, has few adjustable parameters, and is easy to parallelize and scale. 
     Most existing LDA implementations are built upon MPI or Map/Reduce that read/write data from/to file systems, including local file systems, networked file systems, and distributed file systems like a Hadoop distributed file system (HDFS). LDA has a large memory requirement since it is necessary to aggregate results in a memory for processing. MPI and Map/Reduce are batch processing systems, and, as such, they can manipulate memory to meet the memory requirements without disrupting other ongoing processing tasks. This is not true for relational databases. There are no in-database SQL-like implementations of LDA for relational databases (RDBMS), particularly not for large distributed shared-nothing MPP databases. In contrast to reading and writing data in file systems, databases read and write data in parallel in tables using queries, which should not consume too much memory. Furthermore, Hadoop and other batch processing systems have parallel mechanisms that are different from those of databases, and batch processing implementations of LDA for file systems are not readily adaptable to databases. 
     It is desirable to provide scalable memory efficient parallel LDA implementations in shared-nothing MPP databases to enable in-database topic modeling and topic-based data analytics, and it is to these ends that the present invention is directed. 
    
    
     
       BRIEF DESCRIPTION OF THE DRAWINGS 
         FIG. 1  is a diagrammatic view of the architecture of a distributed shared-nothing MPP database system with which the invention may be used; 
         FIG. 2  illustrates a preferred embodiment of a node of the database system of  FIG. 1 ; 
         FIG. 3  is a block diagram of an architecture of a software library of database objects for in-database analytics which may be used to implement LDA; 
         FIG. 4  is a diagrammatic view of an example of a workflow of a LDA Gibbs topic modeling process in accordance with the invention; 
         FIG. 5  is an example of a portion of an input work table in accordance with the invention; and 
         FIG. 6 , comprising  FIGS. 6A and 6B , is an example of a portion of a work table in accordance with the invention. 
     
    
    
     DESCRIPTION OF PREFERRED EMBODIMENTS 
       FIG. 1  is an overview of the architecture of one type of distributed relational database system  100  with which the invention may be employed, the database system shown being a massively parallel processing (MPP) shared nothing distributed database system. The database system may include a master node  102  which connects to a cluster of a plurality of distributed segment nodes  104 _A through  104 _N via a network interconnect  106 . Each segment node may comprise one or more database (DB) instances. The master and segment nodes may comprise generally similar server systems and may have similar architectures and process models. Data stored in the database is distributed across and stored on local database instances of the segment nodes. The segment nodes operate together under the direction of the master node to process user data requests (queries) and return results, and to perform other processing operations, such as data analytics, as will be described. The segments  104 _ 1 - 104 _N work together under the direction of the master node  102  to process workloads. 
       FIG. 2  illustrates an example of the architecture of a master node  202  of the database system  100  that is configured to perform processes and operations in accordance with the invention. The master node and the segment nodes may have substantially similar architectures, as stated above. The master node  202  may comprise a host computer server system which may comprise a single CPU  210  or which may be a multi-processor system comprising a plurality of CPUs connected to input/output (I/O) devices  212  by a communications bus  214 . The I/O devices may be standard computer system input and output devices. A network interface module  216  may also be connected to bus  214  to allow the master node to operate in the networked environment. The master node may further have storage  220  comprising non-transitory physical storage media connected to the bus that embodies executable instructions, such as an operating system and application programs, to control the operations of the computer system. Storage  220  may include a main memory  222  embodying control programs that control the CPU to operate in accordance with the invention, as will be described, and may contain other storage including a data store  224  that may store system configuration information, a catalog, logs, applications and utilities, and, as will be described, a software library for in-database analytics processing objects and user-defined functions. 
     The master node  202  may be responsible for accepting queries in SQL or another database structured query language from a client (user), planning a query, slicing a query plan into multiple slices, dispatching the query plan slices to various segments for execution on the locally stored data in the database storage of each segment, and collecting the query results from the segments. The master may also accept directions from a user or other application programs to perform other data analytics processing operations and the like, including LDA processing, as will be described. In addition to interfacing the segment hosts to the master host, the network interface module  216  may also communicate data, instructions and results between execution processes on the master and the segments. 
       FIG. 3  is a block diagram that illustrates the architecture of a software library  310  comprising objects  314  such as user-defined functions (UDFs), user-defined aggregates (UDAs), etc., which are referred to herein as MADlib objects, for in-database analytics. A separate software object library is preferably integrated into the database  300  of each segment node, so that the objects are readily available on all segments. The objects  314  may comprise MADlib processing algorithms  320 , including modules for LDA Gibbs sampling, written in a programming language such as Python that interfaces with the database SQL, and may further comprise a MADlib shared object library  322 . An object may be called by the database kernel on each segment node when the computer on the segment node is executing a query which explicitly or implicitly depends upon the called object&#39;s functionalities. In accordance with the invention, the LDA algorithm may be implemented in a data-parallel way where each segment node has only a part of the dataset and the LDA modeling process is done in parallel on the cluster of nodes, as described more fully below. 
     Prior to describing the invention, Gibbs sampling for LDA will first be described. As explained above, LDA is used to learn characteristics of a dataset to develop a model for inference. The characteristics may include P(w|z), the word distribution for a given topic, and P(z|d), the topic distribution for a given document. Simply stated, Gibbs sampling for LDA informs how to sample (assign) a new topic for a word in a document based on the current topic assignments of the words in a corpus. This requires calculating P(z i =k|z −i , w), the conditional probability distribution of assigning topic k to the i th  word given z −i , the current assignments of topics to all the other words excluding the i th  word. Once the probability distribution is determined, the sampling becomes straightforward. The conditional probability distribution may be determined from the following Equation 1, which indicates how to sample (assign) a new topic for each word in a corpus. Equation 1: 
     
       
         
           
             
               P 
               ⁡ 
               
                 ( 
                 
                   
                     
                       z 
                       i 
                     
                     = 
                     
                       k 
                       | 
                       
                         z 
                         
                           - 
                           i 
                         
                       
                     
                   
                   , 
                   w 
                 
                 ) 
               
             
             ∝ 
             
               
                 
                   
                     n 
                     
                       
                         - 
                         i 
                       
                       , 
                       k 
                     
                     
                       w 
                       i 
                     
                   
                   + 
                   β 
                 
                 
                   
                     n 
                     
                       
                         - 
                         i 
                       
                       , 
                       k 
                     
                   
                   + 
                   
                     W 
                     ⁢ 
                     
                         
                     
                     ⁢ 
                     β 
                   
                 
               
               ⨯ 
               
                 ( 
                 
                   
                     n 
                     
                       
                         - 
                         i 
                       
                       , 
                       k 
                     
                     
                       d 
                       i 
                     
                   
                   + 
                   α 
                 
                 ) 
               
             
           
         
       
     
     The following table gives the meaning associated with each element of Equation 1: 
     
       
         
               
               
             
           
               
                   
               
               
                 Element 
                 Meaning 
               
               
                   
               
             
             
               
                 i 
                 Index of the i th  word in the corpus, i ∈ {1, 2, . . . , M}, 
               
               
                   
                 where M is the total number of words in the corpus 
               
               
                 k 
                 Index of a topic, k ∈ {1, 2, . . . , 7}, where T is the number 
               
               
                   
                 of topics 
               
               
                 w 
                 Sequence of words appearing in the corpus, where |w| = m 
               
               
                 w i   
                 Word identifier of the i th  word, where w i  ∈ {1, 2, . . . , W}, 
               
               
                   
                 where W is the size of vocabulary 
               
               
                 d i   
                 Document identifier of the i th  word, where d i  ∈ {1, 2, . . . , D}, 
               
               
                   
                 where D is the size of corpus 
               
               
                 z i   
                 Topic assigned to the i th  word 
               
               
                 z −i   
                 Topic assignments of all other words except for the i th  word 
               
               
                   
               
               
                 n −i,k   w     i     
                 Count of word w i  assigned to topic k, not including the 
               
               
                   
                 current assignment of the i th  word, i.e. 
               
               
                   
                 
                   
                     
                       
                         
                           n 
                           
                             
                               - 
                               i 
                             
                             , 
                             k 
                           
                           
                             w 
                             i 
                           
                         
                         = 
                         
                           { 
                           
                             
                               
                                 
                                   n 
                                   k 
                                   
                                     w 
                                     i 
                                   
                                 
                               
                               
                                 
                                   
                                     if 
                                     ⁢ 
                                     
                                         
                                     
                                     ⁢ 
                                     
                                       z 
                                       i 
                                     
                                   
                                   ≠ 
                                   k 
                                 
                               
                             
                             
                               
                                 
                                   
                                     n 
                                     k 
                                     
                                       w 
                                       i 
                                     
                                   
                                   - 
                                   1 
                                 
                               
                               
                                 
                                   
                                     if 
                                     ⁢ 
                                     
                                         
                                     
                                     ⁢ 
                                     
                                       z 
                                       i 
                                     
                                   
                                   = 
                                   k 
                                 
                               
                             
                           
                         
                       
                     
                   
                 
               
               
                   
               
               
                 n −i,k   d     i     
                 Count of words assigned to topic k in document d i , 
               
               
                   
                 not including the current assignment of the i th  word, i.e. 
               
               
                   
                 
                   
                     
                       
                         
                           n 
                           
                             
                               - 
                               i 
                             
                             , 
                             k 
                           
                           
                             d 
                             i 
                           
                         
                         = 
                         
                           { 
                           
                             
                               
                                 
                                   n 
                                   k 
                                   
                                     d 
                                     i 
                                   
                                 
                               
                               
                                 
                                   
                                     if 
                                     ⁢ 
                                     
                                         
                                     
                                     ⁢ 
                                     
                                       z 
                                       i 
                                     
                                   
                                   ≠ 
                                   k 
                                 
                               
                             
                             
                               
                                 
                                   
                                     n 
                                     k 
                                     
                                       d 
                                       i 
                                     
                                   
                                   - 
                                   1 
                                 
                               
                               
                                 
                                   
                                     if 
                                     ⁢ 
                                     
                                         
                                     
                                     ⁢ 
                                     
                                       z 
                                       i 
                                     
                                   
                                   = 
                                   k 
                                 
                               
                             
                           
                         
                       
                     
                   
                 
               
               
                 n −i,k   
                 Count of words assigned to topic k in whole corpus, 
               
               
                   
                 not including the current assignment of the i th  word, i.e. 
               
               
                   
                 
                   
                     
                       
                         
                           n 
                           
                             
                               - 
                               i 
                             
                             , 
                             k 
                           
                         
                         = 
                         
                           { 
                           
                             
                               
                                 
                                   n 
                                   k 
                                 
                               
                               
                                 
                                   
                                     if 
                                     ⁢ 
                                     
                                         
                                     
                                     ⁢ 
                                     
                                       z 
                                       i 
                                     
                                   
                                   ≠ 
                                   k 
                                 
                               
                             
                             
                               
                                 
                                   
                                     n 
                                     k 
                                   
                                   - 
                                   1 
                                 
                               
                               
                                 
                                   
                                     if 
                                     ⁢ 
                                     
                                         
                                     
                                     ⁢ 
                                     
                                       z 
                                       i 
                                     
                                   
                                   = 
                                   k 
                                 
                               
                             
                           
                         
                       
                     
                   
                 
               
               
                 n D×T   
                 Per-document topic count matrix of D by T, n z   d  ∈ n D×T   
               
               
                 n W×T   
                 Per-word topic count matrix of W by T, n z   w  ∈ n W×T   
               
               
                 n 1×T   
                 Corpus-level topic count matrix of 1 by T, n z  ∈ n 1×T   
               
               
                   
               
             
          
         
       
     
     As a workflow to calculating the probability distribution of assigning a topic to a word according to Equation 1, initially each word in a document may be randomly assigned a topic. Next, the per-document, per-word and corpus level topic counts according to the random assignment may be determined. These per-document counts may be used to compute the probability distribution during a sampling process, and the foregoing steps iterated where during each iteration each word in each document is sampled (assigned) a new topic and the topic counts are recalculated. Each iteration refines the per per-word topic assignments. The iterations may continue until a stop condition is satisfied. The workflow is illustrated in  FIG. 4  and will be described in more detail below. 
     In a centralized implementation, if the per-word and corpus-level topic count matrixes (n W×T  and n 1×T ) are such that the matrices can be held in memory, the dataset can be handled document by document and the results can be merged into the matrices in memory while initializing or sampling a topic for each word. Also, the process can run in a similar way for a distributed parallel implementation in batch processing systems based upon Hadoop or MPI since separate documents can be distributed to a set of processing nodes. Each processing node can handle a subset of documents if the topic count matrixes n W×T  and n 1×T  are synchronized at initialization and at the end of each iteration. 
     However, there are problems implementing parallel LDA in parallel databases. First, a database is designed as an interactive system, and the execution of any single query should not consume too many resources or inhibit the execution of other queries. In particularly, a SQL query should run within a proper memory quotation and should not over consume memory. The need to store the per-word and corpus-level topic count matrixes (n W×T  and n 1×T ) in memory means that with an increase in the vocabulary size and/or the topic number, the memory usage will also increase which can make the system non-scalable. Also, since databases store data in tables and data access and manipulation are done via SQL queries, it is necessary to design the data structures and SQL queries to avoid manipulating large topic count matrixes. 
     The invention addresses these problems by parceling out the documents of a dataset and distributing subsets of documents to a set of segment nodes for processing. Each node will process its subset of the documents to produce per-word and corpus level topic count matrices that can be held in memory locally at each segment node, and such that the matrices can be synchronized at each iteration of the processing algorithm. This enables a highly scalable and memory-efficient solution for parallel LDA in shared-nothing MPP databases. In a preferred embodiment, the data is distributed to a plurality of segment nodes using the available built-in data distribution mechanism of an MPP database, and queries are dispatched to these segment nodes by the master node where they are executed in parallel. This allows each segment node to do Gibbs sampling locally on a subset of the corpus of documents, which significantly reduces the load on the master node and avoids a scalability bottleneck. 
     As will be described more fully, each document may be represented as quadruple comprising &lt;docid, wordcount, words, counts&gt;, where docid is a document identifier; wordcount is the total number of words in the document; words is a list of unique word identifiers in the document, and counts is a list of integers representing the number of occurrences of unique words. This representation enables each document to be stored as a single row having four separate fields by a single segment node instead of being stored as multiple rows by multiple segments. Thus, the whole dataset may be distributed by docid. Gibbs sampling for LDA needs the per-word topic count to calculate the conditional probability according to the foregoing Equation 1. Representing each document as a single row having separate fields for the quadruple &lt;docid, wordcount, words, counts&gt; as described above requires passing only one row to a UDF sampling operation instead of the whole topic count matrix, thereby avoiding over consumption of memory. 
       FIG. 5  is an example of one row of an input data table having four fields (columns): docid; wordcount; words; and counts corresponding to the elements of the quadruple. Each row of the table represents a document identified by docid, and the figure shows one row for an example document, docid 1, which has a wordcount of 1544 corresponding to the total number of words in the document docid 1. The words column of the table is a field of comma-delimited data that gives the list of unique words in docid 1, where each unique word is represented by a unique numerical value. (In  FIG. 5 , the words list shown has been truncated, as indicated by the dots, to reduce the size of the figure.) The counts column is field comprising a list of comma-delimited numbers that represent the number of occurrences of the unique words in the document. The number of unique words is len(words)=len(counts)=531 in the example shown in the figure, and the sum of the counts is equal to the wordcount, i.e., sum(counts)=wordcount. 
     A key part of Gibbs sampling for LDA is the need to update topic assignments of words iteration by iteration, as previously described. The invention may employ data structures in the form of one or more work tables for holding topic assignments. The work tables may have a structure that is similar to one holding the training dataset (corpus), as shown and as will be described in connection with  FIG. 6 . Furthermore, the invention affords an implementation that significantly improves efficiency by enabling new data to be inserted into a table rather than updating existing data in a table. INSERT operations in databases to insert data into tables are more efficient than UPDATE operations that update existing data. In a preferred embodiment, the invention employs separate data structures (tables) and INSERT operations rather than UPDATE operations for updating one table with topic assignments for each iteration of the sampling algorithm. In an embodiment, the invention provides two work tables having the same structure, Table 0 and Table 1, which are used alternately as input and output tables. During a first iteration, the first table, Table 0, may be used as the input to the algorithm and the second table, Table 1, is used as the output of the algorithm. In the next iteration, the roles of the tables will reverse. Table 1 populated from the first iteration will serve as the input to the algorithm for the second iteration, and the results of the sampling algorithm will be inserted into Table 0. This “ping-ponging” of tables significantly reduces the time needed for data manipulation and speeds up the whole sampling process. 
       FIG. 4  is a diagrammatic view of an exemplary workflow of LDA Gibbs topic modeling in accordance with the invention. The workflow may be run in parallel on each separate segment node of the MPP database. Referring to  FIG. 4 , at  402  the topic assignment of each word in a dataset on a node may be first initialized by randomly generating an initial topic assignment to each word in each document. This may be accomplished, for example, by a user defined function (UDF): MADLib.random_assign that randomly generates the initial topic assignment in a first Work Table 0 from an input data table by generating random numbers for topic assignments and inserting them into the Work Table 0. This may be accomplished using, for example, the following SQL script: 
     
       
         
               
             
               
               
             
           
               
                   
               
             
             
               
                 INSERT INTO work_table_0 
               
               
                  SELECT 
               
               
                   docid, wordcount, words, counts, 
               
               
                   MADLib.random_assign(wordcount, topic_num) AS doc_topic 
               
               
                  FROM data_table 
               
               
                   Note that doc_topic is a composite type defined as follows: 
               
               
                    CREATE TYPE doc_topic AS( 
               
             
          
           
               
                     topic_dist 
                 INTEGER[ ], 
               
               
                     topic_assign 
                 INTEGER[ ]) 
               
               
                   
               
             
          
         
       
     
     Returning to  FIG. 4 , in steps  404  and  406  the workflow process obtains the corpus level and per-word topic distributions (models). This may be accomplished with the following SQL script: 
     
       
         
               
               
             
               
               
               
             
           
               
                   
                   
               
             
             
               
                   
                 INSERT INTO model_table 
               
               
                   
                 SELECT 
               
               
                   
                  voc_size, topic_num, alpha, beta, 
               
               
                   
                  MADLib.count_topic_agg ( 
               
               
                   
                    words, counts, 
               
               
                   
                    doc_topic, 
               
               
                   
                   voc_size, topic_num) AS model 
               
               
                   
                 FROM work_table_in 
               
               
                   
                 Note that model is a composite type defined as follows: 
               
               
                   
                   CREATE TYPE model AS( 
               
             
          
           
               
                   
                    corpus_topic_dist 
                 INTEGER[ ], 
               
               
                   
                    word_topic_dist 
                 INTEGER[ ]) 
               
               
                   
                   
               
             
          
         
       
     
     The SQL script obtains the topic distributions for the segments, and includes a UDA operation MADLib.count topic_agg that aggregates word and topic counts from across all database segments to provide word counts, document topics and number and vocabulary size. The doc_topic element may be a composite type that includes the topic distribution topic_dist in a document (the number of words assigned to a topic in a document), and topic_assign is the topic assignment of each word in the document comprising an array of word counts per topic. Different occurrences of the same word in a document may have different topic assignments. 
       FIG. 6  is an example of a portion of one row (for one document) of a working table produced by the iterative Gibbs sampling process or by the random initialization script. The table comprises six columns (fields of the row). The columns correspond to the fields docid, wordcount, words and counts which may be the same and have the same data format as described above and shown in  FIG. 5 . (Similar to  FIG. 5 , the words and the topic_assign columns of the table of  FIG. 6  have been truncated, as indicated by the dots, to reduce the size of the figure.) 
     Steps  408  and  410  of  FIG. 4  illustrate one form of an iteration process for the Gibbs sampling process of Equation 1. As long as the output of step  406  has not reached a stop condition, Gibbs sampling is performed at  410  where the per-document, per-word and corpus-level topic counts are used to sample new topics for each word in each document to compute the probability distribution according to Equation 1. The results may be fed back at  412  to step  404 , and the process repeated. Upon a stop condition being satisfied, e.g., a change of likelihood between two adjacent iterations being smaller that a predetermined threshold, or a predetermined number of iterations being reached, the process branches at  414  to step  416  where which the LDA results from Gibbs sampling may be aggregated and output as models. The models may be stored in the database, and used to situate new input data into existing models. Queries may also be run on the stored database models to derive desired information. 
     The following SQL script may be used for the Gibbs sampling  410 , where Work Table 0 and Work Table 1 may be used alternately as work_table_out and work_table_in: 
     
       
         
               
             
           
               
                   
               
             
             
               
                 INSERT INTO work_table_out 
               
               
                 SELECT 
               
               
                  docid, wordcount, words, counts, 
               
               
                  MADLib.gibbs_sample( 
               
               
                   words, counts, doc_topic, 
               
               
                   (SELECT model FROM model table), 
               
               
                   alpha, beta, voc_size, topic_num) 
               
               
                 FROM 
               
               
                  work_table_in 
               
               
                  Note that gibbs_sample is a stateful UDF which allows the updated 
               
               
                  model to be transferred to the continuous function invocations. 
               
               
                   
               
             
          
         
       
     
       FIG. 6  is an example of Work Tables 0 and 1. The doc_topic term may be obtained directly from the Work Table. The word_topic and corpus_topic terms may be obtained from the function MADLib.count_topic_agg. 
     In the foregoing SQL scripts, there are two UDFs, i.e., random_assign and gibbs_sample, and one UDA, i.e., count_topic_agg, indicated that are constructed to have the following functionality:
         random_assign: assigns a topic to each occurrence of a word in a document randomly and returns an array of topic assignments;   gibbs_sample: computes the conditional probability distribution according to Equation 1, draws a new topic from the distribution, and returns the new topic; and   count_topic_agg: scans through all the topic assignments, accumulates the number of occurrences of each topic, and returns an array of topic counts. The transition state variable (which stores the per-word and corpus-level topic counts may be very large, e.g., 1040 MB, when the vocabulary size*topic_num is large. To reduce the network overhead, the state variable can be compressed, because the per-word topic counts is usually a sparse matrix, before transferring from the segment node to the master node for merging.
 
These UDFs and UDA may be written in any convenient programming language, such as Python, to interface with the SQL code of the scripts.
       

     As may be appreciated from the foregoing, the invention affords a highly-scalable, memory-efficient parallel LDA process for a shared-nothing MPP database using a native SQL-based approach which avoids the necessity of moving data between database tables and file systems. By writing directly to alternate work tables and processing parts of the dataset in parallel on distributed database segments, as described above, the invention does not require a large amount of memory, can readily scale to handle a very big dataset, and can achieve near-linear speedup as the number of database segment nodes increase. 
     While the foregoing has been with respect to preferred embodiments of the invention, it will be appreciated that changes to these embodiments may be made without departing from the principles and spirit of the invention, the scope of which is defined in the appended claims.