Hadoop Bloom filter应用示例

2022-07-12 16:56:05

2014-06-04 11:55 451人阅读 评论(0) 收藏 举报

1. 简介

参见《Hadoop in Action》P102 以及 《Hadoop实战（第2版）》（陆嘉恒）P69

Hadoop Bloom filter应用示例

Hadoop Bloom filter应用示例

2. 案例

网上大部分的说明仅仅是按照《Hadoop in Action》中的示例代码给出，这里是Hadoop0.20.2版本，在该版本中已经实现了BloomFilter。

案例文件如下：

customers.txt

1,Stephanie Leung,555-555-5555
    2,Edward Kim,123-456-7890
    3,Jose Madriz,281-330-8004
    4,David Stork,408-555-0000

-----------------------------------------------------------------

orders.txt

3,A,12.95,02-Jun-2008
    1,B,88.25,20-May-2008
    2,C,32.00,30-Nov-2007
    3,D,25.02,22-Jan-2009
    5,E,34.59,05-Jan-2010
    6,F,28.67,16-Jan-2008
    7,G,49.82,24-Jan-2009

两个文件通过customer ID关联。

3. 代码

  1. import java.io.BufferedReader;
  2. import java.io.IOException;
  3. import java.io.InputStreamReader;
  4. import java.util.ArrayList;
  5. import org.apache.hadoop.conf.Configuration;
  6. import org.apache.hadoop.fs.FSDataInputStream;
  7. import org.apache.hadoop.fs.FileSystem;
  8. import org.apache.hadoop.fs.Path;
  9. import org.apache.hadoop.io.Text;
  10. import org.apache.hadoop.mapreduce.Job;
  11. import org.apache.hadoop.mapreduce.Mapper;
  12. import org.apache.hadoop.mapreduce.Reducer;
  13. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  14. import org.apache.hadoop.mapreduce.lib.input.FileSplit;
  15. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
  16. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  17. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
  18. import org.apache.hadoop.util.GenericOptionsParser;
  19. import org.apache.hadoop.util.bloom.BloomFilter;
  20. import org.apache.hadoop.util.bloom.Key;
  21. import org.apache.hadoop.util.hash.Hash;
  22. public class BloomMRMain {
  23. public static class BloomMapper extends Mapper<Object, Text, Text, Text> {
  24. BloomFilter bloomFilter = new BloomFilter(10000, 6, Hash.MURMUR_HASH);
  25. protected void setup(Context context) throws IOException ,InterruptedException {
  26. Configuration conf = context.getConfiguration();
  27. String path = "hdfs://localhost:9000/user/hezhixue/input/customers.txt";
  28. Path file = new Path(path);
  29. FileSystem hdfs = FileSystem.get(conf);
  30. FSDataInputStream dis = hdfs.open(file);
  31. BufferedReader reader = new BufferedReader(new InputStreamReader(dis));
  32. String temp;
  33. while ((temp = reader.readLine()) != null) {
  34. //              System.out.println("bloom filter temp:" + temp);
  35. String[] tokens = temp.split(",");
  36. if (tokens.length > 0) {
  37. bloomFilter.add(new Key(tokens[0].getBytes()));
  38. }
  39. }
  40. }
  41. protected void map(Object key, Text value, Context context) throws IOException ,InterruptedException {
  42. //获得文件输入路径
  43. String pathName = ((FileSplit) context.getInputSplit()).getPath().toString();
  44. if (pathName.contains("customers")) {
  45. String data = value.toString();
  46. String[] tokens = data.split(",");
  47. if (tokens.length == 3) {
  48. String outKey = tokens[0];
  49. String outVal = "0" + ":" + tokens[1] + "," + tokens[2];
  50. context.write(new Text(outKey), new Text(outVal));
  51. }
  52. } else if (pathName.contains("orders")) {
  53. String data = value.toString();
  54. String[] tokens = data.split(",");
  55. if (tokens.length == 4) {
  56. String outKey = tokens[0];
  57. System.out.println("in map and outKey:" + outKey);
  58. if (bloomFilter.membershipTest(new Key(outKey.getBytes()))) {
  59. String outVal = "1" + ":" + tokens[1] + "," + tokens[2]+ "," + tokens[3];
  60. context.write(new Text(outKey), new Text(outVal));
  61. }
  62. }
  63. }
  64. }
  65. }
  66. public static class BloomReducer extends Reducer<Text, Text, Text, Text> {
  67. ArrayList<Text> leftTable = new ArrayList<Text>();
  68. ArrayList<Text> rightTable = new ArrayList<Text>();
  69. protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException ,InterruptedException {
  70. leftTable.clear();
  71. rightTable.clear();
  72. for (Text val : values) {
  73. String outVal = val.toString();
  74. System.out.println("key: " + key.toString() + " : " + outVal);
  75. int index = outVal.indexOf(":");
  76. String flag = outVal.substring(0, index);
  77. if ("0".equals(flag)) {
  78. leftTable.add(new Text(outVal.substring(index+1)));
  79. } else if ("1".equals(flag)) {
  80. rightTable.add(new Text(outVal.substring(index + 1)));
  81. }
  82. }
  83. if (leftTable.size() > 0 && rightTable.size() > 0) {
  84. for(Text left : leftTable) {
  85. for (Text right : rightTable) {
  86. context.write(key, new Text(left.toString() + "," + right.toString()));
  87. }
  88. }
  89. }
  90. }
  91. }
  92. public static void main(String[] args) throws Exception {
  93. Configuration conf = new Configuration();
  94. String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
  95. if (otherArgs.length != 2) {
  96. System.err.println("Usage: BloomMRMain <in> <out>");
  97. System.exit(2);
  98. }
  99. Job job = new Job(conf, "BloomMRMain");
  100. job.setJarByClass(BloomMRMain.class);
  101. job.setMapperClass(BloomMapper.class);
  102. job.setReducerClass(BloomReducer.class);
  103. job.setInputFormatClass(TextInputFormat.class);
  104. job.setOutputFormatClass(TextOutputFormat.class);
  105. job.setMapOutputKeyClass(Text.class);
  106. job.setMapOutputValueClass(Text.class);
  107. job.setOutputKeyClass(Text.class);
  108. job.setOutputValueClass(Text.class);
  109. FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
  110. FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
  111. System.exit(job.waitForCompletion(true) ? 0 : 1);
  112. }
  113. }
上一篇：js 删除字符串中所有空格


下一篇：H5 类选择器