Showing posts with label parallel programming. Show all posts
Showing posts with label parallel programming. Show all posts

Friday, December 14, 2012

Apriori in Java (Part 2)


(Part 1)

Apriori Algorithm:

The algorithm works as follows: first it generates all the frequent itemsets of length 1 w.r.t. the given threshold (minimum support). Then it continue to generate itemsets of lengths 2,3, ... ,n if possible. There are lots of improvements and pruning possible in the implementation.

Do we need to sort frequently? First observation is, if the items are taken in sorted order on step k = 1, all the frequent patterns generated in future will also be in sorted order if maintained properly. So this eliminates the necessity of sorting or storing itemsets in logarithmic data structures like map or set. Rather, we can store them in arraylist or vector like data structure.

How do we generate itemsets of length k+1? We generate itemsets of length k+1 by merging two itemsets of length k. If two itemsets I1 and I2 have a common prefix of length k-1, and I1[k] < I2[k], we can take I1[1 ... k-1] I1[k] I2[k] which is an itemset of length k+1. As our itemsets are sorted, and frequent itemsets generated are also sorted, this can be done in O(N*K^2). Well, if we follow the naive approach, it will take O(N*K^3), however, as we can pre-calculate the length of common prefix of consecutive items in O(N*K), later, we can use this to perform joining operation stated above and also do early termination rather than looking for the entire N items. The approach is demonstrated in source code.

What are the prunings? The most important observation on apriori algorithm is, if an itemset is not frequent, none of its superset can be frequent. Which also tells us, if an itemset has to be frequent, all of its subset must be frequent. Well, the first one is automatically checked in the algorithm. As it takes the frequent itemsets of previous step to calculate itemsets of current step. Now, how the second rule is checked? Well, we don't need to check all the subsets really, all we need is to check the subsets of length k-1 for step k, which can be easily done by performing manual hashing, or using java hashmap. This saves a lots of hassles.

/*
Author: Zobayer Hasan
*/
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;

public class Apriori extends Thread {
    public static boolean debugger = false;
    
    private final Database db;
    private final List< Integer > itemset;
    private final List< List< Integer > > frequent;
    private double minsup;

    public Apriori(String thrName, Database db, double minsup) {
        super(thrName);
        this.db = db;
        itemset = db.getItemset();
        frequent = new ArrayList< List< Integer > >();
        this.minsup = minsup;
    }
    
    @Override
    public void run() {
        double startTime = System.currentTimeMillis();
        
        int k = 1, n = db.dbSize();
        List< List< Integer > > Ck = new ArrayList< List< Integer > >();
        List< List< Integer > > Lk = new ArrayList< List< Integer > >();
        HashMap< List< Integer>, Integer > seenK = new HashMap< List< Integer >, Integer >();
        
        for(Integer item : itemset) {
            List< Integer > temp = new ArrayList< Integer >();
            temp.add(item);
            Lk.add(temp);
        }
        
        while(k <= n && !Lk.isEmpty()) {
            if(debugger) {
                System.out.println("Step " + k);
                System.out.println("Lk: " + Lk);
            }
            
            seenK.clear();
            Ck.clear();
            for(List< Integer > kth : Lk) {
                int count = db.scanDatabase(kth);
                if((double)count < Math.ceil(minsup * (double)n / 100.0)) continue;
                Ck.add(kth);
            }
            
            if(debugger) {
                System.out.println("Ck: " + Ck);
            }
            
            if(Ck.isEmpty()) break;
            
            for(List< Integer > freq : Ck) {
                frequent.add(freq);
                seenK.put(freq, k);
            }
            
            int[] prefixlen = new int[Ck.size()];
            prefixlen[0] = 0;
            for(int i = 1; i < Ck.size(); i++) {
                prefixlen[i] = prefixLen(Ck.get(i-1), Ck.get(i));
            }
            
            List< List< Integer > > temp = new ArrayList< List< Integer > >();
            for(int i = 0; i < Ck.size(); i++) {
                for(int j = i + 1; j < Ck.size(); j++) {
                    if(prefixlen[j] == k-1) {
                        if(debugger) {
                            System.out.println("Joining: " + i + ":" + Ck.get(i) + " + " + j + ":" + Ck.get(j) + " Prefix Length " + prefixlen[j]);
                        }
                        temp.add(prefixJoin(Ck.get(i), Ck.get(j)));
                    }
                    else break;
                }
            }
            
            if(debugger) {
                System.out.println("Temporary: " + temp);
            }

            Lk.clear();
            for(List< Integer > list : temp) {
                boolean candid = true;
                if(k > 1) {
                    for(int i = 0; i < list.size(); i++) {
                        List< Integer > prev = new ArrayList< Integer >();
                        for(int j = 0; j < list.size(); j++) {
                            if(i != j) prev.add(list.get(j));
                        }
                        if(!seenK.containsKey(prev)) {
                            candid = false;
                            break;
                        }
                    }
                }
                if(candid) {
                    Lk.add(list);
                }
            }
            
            if(debugger) {
                System.out.println("Pruned: " + Lk);
            }
            
            k++;
        }
        
        double endTime = System.currentTimeMillis();
        System.out.println("Apriori completed in " + (endTime - startTime)/1000.0 + " seconds");
    }
    
    public void printPatterns() {
        System.out.println("Frequent Itemsets");
        for(List< Integer > pattern : frequent) {
            System.out.println(pattern);
        }
        System.out.println("Total " + frequent.size() + " itemsets");
    }
    
    private int prefixLen(List< Integer > left, List< Integer > right) {
        int len = 0;
        for(len = 0; len < left.size() && len < right.size(); len++) {
            if(left.get(len).compareTo(right.get(len)) != 0) return len;
        }
        return len;
    }
    
    private List< Integer > prefixJoin(List< Integer > left, List< Integer > right) {
        List< Integer > ret = new ArrayList< Integer >();
        for(Integer i : left) {
            ret.add(i);
        }
        ret.add(right.get(right.size() - 1));
        return ret;
    }
}
This class is threaded, so it is possible to take advantages of multicore processors.

A sample test class is shown here.
/*
Author: Zobayer Hasan
*/
public class FIMtest {

    public static void main(String[] args) {
        Database db = null;
        try {
            db = new Database("mushroom.dat");
            
        } catch(Exception e) {
            e.printStackTrace();
        }
        
        System.out.println("\nStarting Apriori");
        
        Apriori test1 = new Apriori("test1", db, 40.0);
        Apriori.debugger = true;
        test1.start();
        try {
            test1.join();
            test1.printPatterns();
        } catch(Exception e) {
            e.printStackTrace();
        }
    }
}

So, this is it, an efficient implementation of Apriori algorithm in java. It is possible that there are some lacking or mistakes in either source code or analysis. If you find any, please let me know. Here is a sample run on mushroom for 40% minimum support.

Have fun...

Thursday, December 13, 2012

Apriori in Java (Part 1)


Introduction:

Apriori is a very basic and straight forward algorithm for frequent pattern mining, I will not be discussing much about the approach, as those can already be studied from different lectures/books available on net. I will basically present an implementation of mine which is an efficient implementation of the standard apriori algorithm in Java. The target input files are taken from Frequent Itemset Mining Dataset Repository, for example, the mushroom dataset. Due to the space and runtime complexity, apriori is not suitable for larger files having 50,000 records or so. It also may take huge time for very dense files.

The Database:

First we need to read the files and store them in some data structure which will be easy to access and efficient at the same time. Here. I am presenting my java source code, and then I will present an explaining analysis why this works better.
/*
Author: Zobayer Hasan
*/

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.PriorityQueue;

class Entry {
    public Integer first;
    public Integer second;
    Entry() {}
    Entry(Integer first, Integer second) {
        this.first = first;
        this.second = second;
    }
}

public class Database {
    public static boolean debugger = false;
    
    private final List< List< Integer > > transactions;
    private final List< Integer > items;

    public Database(String dataFileName) throws Exception {
        if(debugger) {
            System.out.println("Processing " + dataFileName);
        }
        
        transactions = new ArrayList< List< Integer > >();
        items = new ArrayList< Integer >();
        
        FileInputStream fin = new FileInputStream(dataFileName);
        InputStreamReader istream = new InputStreamReader(fin);
        BufferedReader stdin = new BufferedReader(istream);
        
        String line;
        
        double startTime = System.currentTimeMillis();
        
        while((line = stdin.readLine()) != null) {
            List< Integer > transaction = new ArrayList< Integer >();
            String[] temp = line.split("\\s+");
            
            for(String num : temp) {
                transaction.add(Integer.parseInt(num));
            }
            
            if(transaction.isEmpty()) continue;
            
            Collections.sort(transaction);
            transactions.add(transaction);
        }
        
        fin.close();
        istream.close();
        stdin.close();
        
        int n = transactions.size();
        int[] header = new int[n];
        PriorityQueue< Entry > pQ = new PriorityQueue< Entry >(n, new Comparator< Entry >() {
            public int compare(Entry item1, Entry item2) {
                if(item1.first.equals(item2.first)) {
                    return item1.second.compareTo(item2.second);
                } else {
                    return item1.first.compareTo(item2.first);
                }
            }
        });
        
        for(int i = 0; i < n; i++) {
            header[i] = 0;
            pQ.add(new Entry(transactions.get(i).get(header[i]), i));
        }
        
        while(!pQ.isEmpty()) {
            Entry peek = pQ.remove();
            int val = peek.first;
            int idx = peek.second;
            if(items.isEmpty() || items.get(items.size()-1) < val) {
                items.add(val);
            }
            while(header[idx] < transactions.get(idx).size() && transactions.get(idx).get(header[idx]) <= val) {
                header[idx]++;
            }
            if(header[idx] < transactions.get(idx).size()) {
                pQ.add(new Entry(transactions.get(idx).get(header[idx]), idx));
            }
        }
        
        double endTime = System.currentTimeMillis();
        System.out.println("Database created in " + (endTime - startTime)/1000.0 + " seconds");
    }
    
    public int scanDatabase(List< Integer > transaction) {
        int count = 0;
        for(List< Integer > row : transactions) {
            boolean found = true;
            for(Integer item : transaction) {
                int idx, stp, st = 0, en = row.size(), cnt = en - st;
                while(cnt > 0) {
                    stp = cnt >> 1; idx = st + stp;
                    if(row.get(idx).compareTo(item) < 0) {
                        st = ++idx;
                        cnt -= stp+1;
                    }
                    else {
                        cnt = stp;
                    }
                }
                if(st == row.size() || row.get(st).compareTo(item) != 0) {
                    found = false;
                    break;
                }
            }
            if(found) count++;
        }
        return count;
    }
    
    public List< Integer > getItemset() {
        return items;
    }
    
    public int dbSize() {
        return transactions.size();
    }
    
    public List< Integer > getRow(int row) {
        try {
            return transactions.get(row);
        } catch(Exception e) {
            throw e;
        }
    }
}
Clearly it provides some interface to access the database where the constructor must be called with a file name for example 'mushroom.dat'. The sorting on line 56 is not necessary if you know the transactions in the file will be sorted in ascending order. All the data files in this repository are already sorted, so sorting can be disabled.

Now, if we look at the constructor, what is this huge code doing actually? First it treads each transaction as a string, then parses it and inserts the transaction in a sorted list. Now, we need a list of unique elements. Well, this could be done by sorting all the records together, and then eliminating duplicates. However, this can be performed better with the help of a priority queue in O(NK log N) time, where we have n transactions, and each transaction has K records on an average. Actually this works much better than it would be in naive approach which would take O(NK log NK) because of extra sorting overhead.

Then comes the last important part, scanDatabase() method. Which basically searches each record in the database for a set of elements. It takes O(NK log K) time at most, where, we have N records, each having an average length of K. This is much better than looking in O(N*K^2) following a naive approach. Here is some improvement possible, if we know the length of transactions, we could use a bit vector in addition to each transaction. Then the query can be performed in O(NK). However, as the datasets are huge, I didn't go for it due to the necessity of using huge memory.

Part 2 contains the source code and explanation for the apriori algorithm.

Wednesday, September 29, 2010

Threaded Merge Sort



What?


In my previous post about Merge Sort, the algorithm for merge sort was discussed. Although the pictures shows something which appears to be apparently parallel, but actually they are not. In fact, as the procedure introduced there was recursive, it works in a post-order fashion, meaning, first it solves the left half, then the right half, and then it works with the current range. The following picture shows the actual work-flow of the standard recursive merge sort algorithm:


The red lines and numbers shows the actual call sequence of the algorithm merge sort, for reference, the basic algorithm, is shown here again:

MERGE-SORT(A, p, r):
if p < r
then q := (r + p) / 2
MERGE-SORT(A, p, q)
MERGE-SORT(A, q+1, r)
MERGE(A, p, q, r)

So, from the above example, we can see that no two calls are concurrent, i.e. parallel. Each call is one of the three calls made in MERGE-SORT(A, p, r) procedure stated above, also shown below (according to the picture, p=0, r=6):

Step-00: MERGE-SORT(A, 0, 6)
Step-01: MERGE-SORT(A, 0, 3)
Step-02: MERGE-SORT(A, 0, 1)
Step-03: MERGE-SORT(A, 0, 0)
Step-04: MERGE-SORT(A, 1, 1)
Step-05: MERGE(A, 0, 0, 1)
Step-06: MERGE-SORT(A, 2, 3)
Step-07: MERGE-SORT(A, 2, 2)
Step-08: MERGE-SORT(A, 3, 3)
Step-09: MERGE(A, 2, 2, 3)
Step-10: MERGE(A, 0, 1, 3)
Step-11: MERGE-SORT(A, 4, 6)
Step-12: MERGE-SORT(A, 4, 5)
Step-13: MERGE-SORT(A, 4, 4)
Step-14: MERGE-SORT(A, 5, 5)
Step-15: MERGE(A, 4, 4, 5)
Step-16: MERGE-SORT(A, 6, 6)
Step-17: MERGE(A, 4, 5, 6)
Step-18: MERGE(A, 0, 3, 6)

Also note that, there are no overlapping calls at a same level of the tree, which is going to be a key factor.

How threading helps:


After a few observations, we can easily find out some interesting properties about merge sort algorithm. If we closely look at the merge sort tree structure, it becomes very clear that, int the recursive call tree, parent nodes depend on children, but siblings are independent. This property helps us to deduce a threaded version of merge sort algorithm where, we can really solve the left and right sub-portion of a range in a parallel fashion by making the calls threaded.

As the algorithm here is recursive, it might look confusing at the first glance, but it is not. Because, nodes share data only with their ancestors and in a call, we can wait until the threads finish their works. So, no concurrency problem appears here. Also, we don't need to worry about mutual exclusion of the shared data, because, there is no such situation in this algorithm, already stated above.

So, the threaded version of the algorithm will look like this:

THREADED-MERGE-SORT(A, p, r):
if p < r
then q := (r + p) / 2
start_thread(left_thread, THREADED-MERGE-SORT(A, p, q))
start_thread(right_thread, THREADED-MERGE-SORT(A, q+1, r))
wait_for(left_thread)
wait_for(right_thread)
MERGE(A, p, q, r)

According to the above algorithm, the flow of program changes which is shown in the picture below:


Now because of being able to run multiple THREADED-MERGE-SORT() at a time, and the job scheduling features of modern computers, the effective number of iteration is dramatically reduced, resulting a much much elegant solution for the heavy duty areas like manipulating large data on drives.

A simple implementation:


Here a simple implementation is presented using pthread library in C++:

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>

typedef struct { int i, j; } pair;
const int MAX = 1024;
int a[MAX];
pthread_attr_t attr;

void *threaded_merge_sort(void *param) {
pair range = *(pair *)param, lrange, rrange;
pthread_t lside, rside;
int p = range.i, q, r = range.j, n1, n2, n = r-p+1, i, j, k;
int *aleft, *aright;
if(p < r) {
q = (p + r) >> 1;
lrange.i = p, lrange.j = q, rrange.i = q + 1, rrange.j = r;
pthread_create(&lside, &attr, threaded_merge_sort, (void *)&lrange);
pthread_create(&rside, &attr, threaded_merge_sort, (void *)&rrange);
pthread_join(lside, NULL);
pthread_join(rside, NULL);
n1 = q - p + 1, n2 = r - q;
aleft = (int *)malloc(sizeof(int) * n1);
aright = (int *)malloc(sizeof(int) * n2);
for(i = 0; i < n1; i++) aleft[i] = a[p+i];
for(i = 0; i < n2; i++) aright[i] = a[q+1+i];
for(k = i = j = 0; k < n; k++) {
if(i >= n1 || (j < n2 && aleft[i] > aright[j])) a[k+p] = aright[j++];
else a[k+p] = aleft[i++];
}
free(aleft);
free(aright);
}
return NULL;
}

int main() {
int n, i;
pthread_t sorter;
pair range;
pthread_attr_init(&attr);
pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
while(scanf("%d", &n)==1 && n) {
for(i = 0; i < n; i++) scanf("%d", &a[i]);
range.i = 0, range.j = n-1;
pthread_create(&sorter, &attr, threaded_merge_sort, (void *)&range);
pthread_join(sorter, NULL);
for(i = 0; i < n; i++) printf("%d%c", a[i], (i==n-1? '\n' : ' '));
}
pthread_attr_destroy(&attr);
return 0;
}

This may look a bit messy, but, actually, you may find a little difference with the original one, we just created threads instead of plain recursive calls, that's it.
To learn more about pthread, check these out:

Give it a try! :)