mahout使用PFP和FPG算法

mahout提供了内存中的FPG和分布式的PFP两种算频繁项集的方法，其中PFP实现上也是将feature分组，然后在节点上独立地运行FPG算法。PFP默认分组为50，如果项的数量特别多，可能需要考虑修改这个值。

先来看一下mahout 0.5的FPG测试代码：

  public void testMaxHeapFPGrowth() throws Exception {FPGrowth<String> fp = new FPGrowth<String>();Collection<Pair<List<String>,Long>> transactions = new ArrayList<Pair<List<String>,Long>>();transactions.add(new Pair<List<String>,Long>(Arrays.asList("E", "A", "D", "B"), 1L));transactions.add(new Pair<List<String>,Long>(Arrays.asList("D", "A", "C", "E", "B"), 1L));transactions.add(new Pair<List<String>,Long>(Arrays.asList("C", "A", "B", "E"), 1L));transactions.add(new Pair<List<String>,Long>(Arrays.asList("B", "A", "D"), 1L));transactions.add(new Pair<List<String>,Long>(Arrays.asList("D"), 1L));transactions.add(new Pair<List<String>,Long>(Arrays.asList("D", "B"), 1L));transactions.add(new Pair<List<String>,Long>(Arrays.asList("A", "D", "E"), 1L));transactions.add(new Pair<List<String>,Long>(Arrays.asList("B", "C"), 1L));Path path = getTestTempFilePath("fpgrowthTest.dat");Configuration conf = new Configuration();FileSystem fs = FileSystem.get(conf);SequenceFile.Writer writer =new SequenceFile.Writer(fs, conf, path, Text.class, TopKStringPatterns.class);fp.generateTopKFrequentPatterns(transactions.iterator(),fp.generateFList(transactions.iterator(), 3),3,100,new HashSet<String>(),new StringOutputConverter(new SequenceFileOutputCollector<Text,TopKStringPatterns>(writer)),new ContextStatusUpdater(null));writer.close();List<Pair<String, TopKStringPatterns>> frequentPatterns = FPGrowth.readFrequentPattern(conf, path);assertEquals("[(C,([B, C],3)), "+ "(E,([A, E],4), ([A, B, E],3), ([A, D, E],3)), "+ "(A,([A],5), ([A, D],4), ([A, E],4), ([A, B],4), ([A, B, E],3), ([A, D, E],3), ([A, B, D],3)), "+ "(D,([D],6), ([B, D],4), ([A, D],4), ([A, D, E],3), ([A, B, D],3)), "+ "(B,([B],6), ([A, B],4), ([B, D],4), ([A, B, D],3), ([A, B, E],3), ([B, C],3))]",frequentPatterns.toString());}

注意这个测试代码的例子，将输入写到了一个sequence file中，这个writer经过StringOutputConverter的包装。在实际跑作业时，这么写是可能出现问题的。因为集群中可能有权限控制，直接用writer往HDFS上写数据，可能造成写入的文件权限出问题（别人都不能读写啦）。

所以如果只是要在控制台看输出，不妨改造一下这个包装类，下面的类实现了在控制台输出频繁项集的结果：

public final class PrintStreamConverter implements OutputCollector<String, List<Pair<List<String>, Long>>> {private final PrintStream collector;public PrintStreamConverter(PrintStream collector) {this.collector = collector;}@Overridepublic void collect(String key,List<Pair<List<String>, Long>> values) throws IOException {for (Pair<List<String>, Long> pair : values) {collector.print(key +": " + StringUtils.join(pair.getFirst(),",") + "\t" + pair.getSecond() + "\n");}
}

注意这里的输出是以feature为key的闭频繁项集及其支持度。

这时FPG代码就可修改为：

 fp.generateTopKFrequentPatterns(transactions.iterator(),fp.generateFList(transactions.iterator(), 3),3,100,new HashSet<String>(),new PrintStreamConverter(System.out),new ContextStatusUpdater(null));

有时候希望在集群节点的内存中做FPG，这时需要一些额外的包装，下面的类提供了包装，并将频繁项集输出为<Text, Text>：

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.mahout.common.Pair;
import org.uncommons.maths.combinatorics.CombinationGenerator;import java.io.IOException;
import java.util.List;public final class TextOutputConverter implements OutputCollector<String, List<Pair<List<String>, Long>>> {private final OutputCollector<Text, Text> collector;public TextOutputConverter(OutputCollector<Text, Text> collector) {this.collector = collector;}@Overridepublic void collect(String key,List<Pair<List<String>, Long>> values) throws IOException {for (Pair<List<String>, Long> pair : values) {collector.collect(new Text(key + "," + StringUtils.join(pair.getFirst(), ";")),new Text(pair.getSecond().toString()));}}
}

reduce中代码如下：

public void reduce(Text key, Iterator<Text> values,OutputCollector<Text, Text> output, Reporter reporter) throws IOException {FPGrowth<String> fp = new FPGrowth<String>();Collection<Pair<List<String>, Long>> transactions = new ArrayList<Pair<List<String>, Long>>();while (values.hasNext()) {List<String> list = new ArrayList<String>();String[] parts = values.next().toString().split(" ");Collections.addAll(list, parts);transactions.add(new Pair<List<String>, Long>(list, 1L));}fp.generateTopKFrequentPatterns(transactions.iterator(),fp.generateFList(transactions.iterator(), 5),5, 1000, new HashSet<String>(),new TextOutputConverter(output),new ContextStatusUpdater(null));
}

FP算法中还有一些可调的参数，通过Parameters类来封装，它是一个<key, value>对集合。

numGroups：feature分组的数目，默认50。对于大项集来说，可能设大一些会好点

input：输入路径

output：输出路径

minSupport：最小支持度，默认为3

mahout使用PFP和FPG算法相关推荐

基于Mahout实现协同过滤推荐算法的电影推荐系统
1 Mahout介绍 Apache Mahout 是 Apache Software Foundation(ASF) 旗下的一个开源项目,提供一些可扩展的机器学习领域经典算法的实现,旨在帮助开发人员更 ...
协同过滤算法_基于Mahout的协同过滤推荐算法
1协同过滤基于物品的协同过滤对物品进行相似度计算,然后再进行推荐. 基于用户的协同过滤基于用户的协同过滤算法是通过用户的历史行为数据发现用户对商品或内容的喜欢,并对这些喜好进行度量和打分.根据不 ...
mahout实现协同过滤推荐算法
前述 mahout支持分类.聚类.推荐等多种功能,这里不展开阐述,只描述协同过滤的使用. 不明白协同过滤是如何计算的同学可以参考这篇文章,通俗易懂. 点击这里 maven依赖 <!-- mhou ...
基于mahout的协同过滤推荐算法实现
1.简述项目架构项目主要实现了基于mahout开源项目实现了协同过滤算法,用java常见的ssm框架搭建,完成了对于用户喜欢的电影进行推荐. 采取MVC模式将业务逻辑.数据.界面显示分离的方法组织代 ...
Mahout推荐算法API详解
Hadoop家族系列文章,主要介绍Hadoop家族产品,常用的项目包括Hadoop, Hive, Pig, HBase, Sqoop, Mahout, Zookeeper, Avro, Ambari, ...
Apache Mahout基于商品的协同过滤算法流程分析
最近使用mahout的itemBase协同过滤算法,研究了下他的源码,记录如下,以备后忘-- 其算法实现大致分四个主要的部分: 1.将输入数据转化成矩阵 2.计算相似性 3.还是转化数据格式,为计算预 ...
mahout贝叶斯算法开发思路（拓展篇）1
首先说明一点,此篇blog解决的问题是就下面的数据如何应用mahout中的贝叶斯算法?(这个问题是在上篇(...完结篇)blog最后留的问题,如果想直接使用该工具,可以在mahout贝叶斯算法拓展下载 ...
《Mahout算法解析与案例实战》一一2.3　测试安装
本节书摘来自华章计算机<Mahout算法解析与案例实战>一书中的第2章,第2.3节,作者:樊哲,更多章节内容可以访问云栖社区"华章计算机"公众号查看. 2.3 测试安 ...
《大数据架构和算法实现之路：电商系统的技术实战》——1.5　相关软件：R和Mahout...
本节书摘来自华章计算机<大数据架构和算法实现之路:电商系统的技术实战>一书中的第1章,第1.5节,作者黄申,更多章节内容可以访问云栖社区"华章计算机"公众号查看. ...

mahout使用PFP和FPG算法

mahout使用PFP和FPG算法相关推荐

最新文章

热门文章