Implementing Apriori Algorithm on Hadoop

Question

I am attempting to implement the Apriori algorithm on using Hadoop. I have already implemented a non-distributed version of the Apriori algorithm but my lack of familiarity with Hadoop and MapReduce has presented a number of concerns.

The way I want to implement the algorithm is in two phases:

1) In the first phase, the map reduce job will operate on the original transaction dataset. The output of this phase is a file containing all of the 1-itemsets and their support of 1.

2) In the second phase, I want to read in the output of the previous phase and then construct the new itemsets. Importantly, I want to then, in the mapper, determine if any of the new itemsets are still found in the dataset. I imagine that if I send the original dataset as the input to the mapper, it will partition the original file so that each mapper only scans through a partial dataset. The candidate list however needs to be constructed from all of the previous phase's output. This will then iterate in a loop for a fixed number of passes.

My problem is figuring out how to specifically ensure that I can access the full itemsets in each mapper, as well as being able to access the original dataset to calculate the new support in each phase.

Thanks for any advice, comments, suggestions or answers.

EDIT: Based on the feedback, I just want to be more specific about what I'm asking here.

Answer 1

Before you start, I suggest you read the Hadoop Map-Reduce Tutorial.

Step 1: Load your data file to HDFS. Let's assume your data is txt file and each set is a line.

a b c
a c d e
a e f
a f z
...

Step 2: Follow the Map-Reduce Tutorial to build your own Apriori Class.

public void map(Object key, Text value, Context context
                ) throws IOException, InterruptedException {
  // Seprate the line into tokens by space
  StringTokenizer itr = new StringTokenizer(value.toString());
  while (itr.hasMoreTokens()) {
    // Add the token into a writable set
    ... put the element into a writable set ...
  }
  context.write(word, one);
}

Step 3: Run the mapreduce jar file. The output will be in a file in the HDFS. You will have something like:

a b 3 (number of occurrence)
a b c 5
a d 2
...

Based on the output file, you could calculate the relationship.

On a related note, you might want to consider using a higher level abstraction than map-reduce like Cascading or Apache Spark.

Answer 2

I implemented AES algorithm in both Apache Spark and Hadoop MapReduce using Hadoop Streaming. I know it is not the same as Apriori but you can try to use my approach.

Simple example of AES implemented using Hadoop Streming MapReduce.

Project structure for AES Hadoop Streaming

1n_reducer.py / 1n_combiner is the same code but without constraint .

import sys

CONSTRAINT = 1000

def do_reduce(word, _values):
    return word, sum(_values)


prev_key = None
values = []

for line in sys.stdin:
    key, value = line.split("\t")
    if key != prev_key and prev_key is not None:
        result_key, result_value = do_reduce(prev_key, values)
        if result_value > CONSTRAINT:
            print(result_key + "\t" + str(result_value))
        values = []
    prev_key = key
    values.append(int(value))

if prev_key is not None:
    result_key, result_value = do_reduce(prev_key, values)
    if result_value > CONSTRAINT:
        print(result_key + "\t" + str(result_value))

base_mapper.py:

import sys


def count_usage():
    for line in sys.stdin:
        elements = line.rstrip("\n").rsplit(",")
        for item in elements:
            print("{item}\t{count}".format(item=item, count=1))


if __name__ == "__main__":
    count_usage()

2n_mapper.py uses the result of previous iteration. In answer to your question, you can read the output of previous iteration to form itemsets in such way.

import itertools
import sys

sys.path.append('.')
N_DIM = 2


def get_2n_items():
    items = set()
    with open("part-00000") as inf:
        for line in inf:
            parts = line.split('\t')
            if len(parts) > 1:
                items.add(parts[0])

    return items


def count_usage_of_2n_items():
    all_items_set = get_2n_items()
    for line in sys.stdin:
        items = line.rstrip("\n").rsplit(",")  # 74743 43355 53554
        exist_in_items = set()
        for item in items:
            if item in all_items_set:
                exist_in_items.add(item)
        for combination in itertools.combinations(exist_in_items, N_DIM):
            combination = sorted(combination)
            print("{el1},{el2}\t{count}".format(el1=combination[0], el2=combination[1], count=1))


if __name__ == "__main__":
    count_usage_of_2n_items()

From my experience, Apriori algorithm is not suitable for Hadoop if the number of unique combinations (items sets) is too large (100K+). If you found an elegant solution for Apriori algorithm implementation using Hadoop MapReduce (Streaming or Java MapReduce implementation) please share with community.

PS. If you need more code snippets please ask for.