本人最近研究Aprior算法，由于要实现海量数据的分析挖掘，需要在hadoop平台加以实现。

在网上看过一些Aprior算法Mapreduce的代码，感觉拿过来都不好直接用，而且，多数都不是原味的Aprior，或者经过改进，是FP-growth算法，或者是将数据分块，各块独立运行Aprior算法，不是严格意义上的Aprior算法。

本人也是几经实验，终于自己也实现了一种基于Mapreduce的原汁原味的Aprior算法的分布式实现。

Aprior分布式实现的问题：输入有多个，一个是事务数据库，一个是全局的候选项集，而候选项集是不能分块的，如果分块交由不同的node执行，分别统计出来的支持度肯定会有缺失。这就要求Aprior算法候选项集要做为一个完整的参数，于是想到了把候选项集存入内存。这就是我这个分布式实现的一个关键点了。

Aprior算法分布式实现算法思路：Mapper：k-v输入为事务数据库，外加一个内存中的数据结构，此数据结构即用来存储候选项集，保证了候选项集是完整的不会被分片。输出为候选项(key)-支持度(value)。Reducer：普通的累加求和即可。在写入文件时加一个判断，满足支持度要求才写入文件。2个基本的原子操作就有了。   

然后一个完整的Aprior算法流程是： 运行了一个job，即一次Mapreduce，产生相应阶次的频繁项集，将此次job的输出读取入内存中，产生高阶候选项集作为下一个job的在内存中的输入，也就是候选项集，然后，启动下一次job。接下来循环迭代，直到不能产生新的频繁集为止。

Mapper类如下：

public static class GetGlobalCandItmSupportMapper extends Mapper<Object, Text,Text, IntWritable>
{//计算出候选项集 输入：key无效,value,事务数据库全部（一次一条效率有些低。产生一个疑问，如果原始文件500m，64m一个分片，一次读入是500m还是64m呢，如果是64m那就很好。后期可以测试一下。）记录，内存中的全局候选项集，输出，Key,一条候选项，value，一条候选项的支持度
//算法概述：获取内存中的候选项集。缺点，每条记录都要进行一次从字符串转候选项集的工作。

int support=3;
private final static IntWritable one = new IntWritable(1);
private Text word = new Text();

public void map(Object key, Text value, Context context) throws IOException, InterruptedException
{
//一个数据结构存储事务数据库，一个数据结构存储所有的候选集，从conf中传入自己想要使用的参数。
ArrayList<ArrayList<String>> rawDB=new ArrayList<>();
ArrayList<ArrayList<String>> CandItms=new ArrayList<>();
String CandItmSets=context.getConfiguration().get("CandItmSets");
System.out.println("CandItmSets are"+CandItmSets);
//构造事务数据库的内存结构

String[] tmpLinesRawDB=value.toString().split("\n"); //获取一行一行的数据
String[] tmpOneLineRawDB=null;
for(int i=0;i<tmpLinesRawDB.length;i++) //一行数据，
{
tmpOneLineRawDB=tmpLinesRawDB[i].split("\\s+");
ArrayList<String> tmpOneLineFrqArr=new ArrayList<>();
for(int j=0;j<tmpOneLineRawDB.length-1;j++) //对一行数据进行分割。 此处实验需要减1了，后续记得加回来
{
tmpOneLineFrqArr.add(tmpOneLineRawDB[j]);
}
if(!tmpOneLineFrqArr.isEmpty())
{
rawDB.add(tmpOneLineFrqArr);
}
}
System.out.println("rawDB is "+rawDB.toString());

//遍历事务数据库与候选项集，进行支持度的计算。
//加一个分支，如果是第一次，即候选项集数组为空，那么统计一元元素的次数，如果候选项集数组不为空，那么统计候选项集中元素出现次数。
if(CandItmSets==null||CandItmSets.isEmpty()||CandItmSets.equalsIgnoreCase("NOFILE")) //为空，说明是第一次，那么统计每个一元元素出现的次数。
{
String[] itr = value.toString().split("\\s+");
for(int i=0;i<itr.length;i++)
{
word.set(itr[i]);
context.write(word, one);
}
}
else//不为空，说明存在候选项集，那么统计候选项集出现的次数。
{
//构造候选项集的内存结构
String[] tmpLines=CandItmSets.split("\n");
String[] tmpOneLine=null;
for(int i=0;i<tmpLines.length;i++)
{
ArrayList<String> tmpLineArr=new ArrayList<>(); //感觉放在里边，可以省略一个克隆对象的操作。
tmpOneLine=tmpLines[i].trim().replace("\\s+", "\\s").split("\\s+");
for(int j=0;j<tmpOneLine.length;j++)
{
tmpLineArr.add(tmpOneLine[j]);
}
CandItms.add(tmpLineArr);
}
//遍历事务数据库，统计候选项集出现次数。
for(int i=0;i<rawDB.size();i++)
{
for(int j=0;j<CandItms.size();j++)
{
if(isArraAContainsArraB(rawDB.get(i), CandItms.get(j)))
{
String arrToString="";
for(int k=0;k<CandItms.get(j).size();k++) // 将字符串数组拼成字符串 arraylist自带的tostring格式不好。
{
arrToString=arrToString.concat(CandItms.get(j).get(k)+" ");
}
System.out.println("radDb Contains frqItm"+"rawdb is"+rawDB.get(i)+"frqItm is"+CandItms.get(j));
context.write(new Text(arrToString), new IntWritable(1));
}
}
}
}
// String[] arr=value.toString().split("\\s+");
// // for(int i=0;i<arr.length)
// if(Integer.valueOf(arr[1])>support)
// context.write(new Text(arr[0]),new IntWritable(Integer.valueOf(arr[1])));
}
}

Reducer类如下：

public static class GetGlobalFrqItmSetsReducer extends Reducer<Text,IntWritable,Text,IntWritable>
{
//一个reducer的输入是一个key的记录，不能把所有输入都在一个reducer中遍历了，所以能做的比较有限。在这里，就只根据支持度要求筛选一下，确定哪些是频繁项集。将频繁项集存入文件。生成高阶候选项集的任务要由mapper来做了。
private IntWritable result = new IntWritable();
public void reduce(Text key, Iterable<IntWritable> values,Context context) throws IOException, InterruptedException
{
ArrayList<ArrayList<String>> frqItmSets=new ArrayList<>();
ArrayList<String> frqOneItm=new ArrayList<>();
ArrayList<String> frqElement=new ArrayList<>();//频繁的元素，给组合算法，产生频繁元素的组合项集。
int sum = 0;
int support=context.getConfiguration().getInt("SUPPORT", 2);
for (IntWritable val : values)
{
sum += val.get();
}
if(sum>support) //大于支持度，才进行一个记录。此处算法要改下，频繁项集要写入内存，下一阶的候选项集才写入文件。
{
String[] tmpOneItm=key.toString().split("\\s+");//将一个频繁项拆成字符串数组
for(int i=0;i<tmpOneItm.length;i++)
{
frqOneItm.add(tmpOneItm[i]);
if(!frqElement.contains(tmpOneItm[i]))
frqElement.add(tmpOneItm[i]);
}
if(!frqOneItm.isEmpty())
{
frqItmSets.add(frqOneItm);//频繁集的数据结构，后期可以删除，此处仅为了看输出的结果是否正确
}
context.write(key,new IntWritable(sum));
}
}
}

Main函数如下

Configuration conf = new Configuration();
conf.setInt("SUPPORT", 3);
conf.set("CandItmSets","NOFILE");
String[] inAndout = new String[]{"hdfs://localhost:9000/bin/in","hdfs://localhost:9000/bin/out0"};
Job job = new Job(conf, "word count");
job.setInputFormatClass(WholeInputFormat.class);
job.setJarByClass(WordCount.class);
job.setMapperClass(GetGlobalCandItmSupportMapper.class);
job.setCombinerClass(GetGlobalFrqItmSetsReducer.class);
job.setReducerClass(GetGlobalFrqItmSetsReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
FileInputFormat.addInputPath(job, new Path(inAndout[0]));
FileOutputFormat.setOutputPath(job, new Path(inAndout[1]));
if(job.waitForCompletion(true))
{
System.out.println("First step has been done");
}

如有疑问，欢迎交流。如有更巧妙的实现方式，敬请指教哈。

Aprior算法Hadoop上实现思路与关键部分代码