如何使用mapreduce避免knn程序溢出？

o8x7eapl  于 2021-06-02  发布在  Hadoop
关注(0)|答案(0)|浏览(262)

我在下面编写的程序产生了大量的溢出（溢出量高达数GB，而我的输入和输出数据只有20mb左右）。
我所做的只是将测试文件存储在缓存中，并在每次将一行列车数据传递到 map() 功能。我无法设置 Combiner 这里是因为每个map（）产生的结果对我的实现来说毫无意义 N = number of test data 记录它们都不共享同一个键（我使用测试数据的索引作为键）。

  1. import java.io.BufferedReader;
  2. import java.io.FileReader;
  3. import java.io.IOException;
  4. import java.util.ArrayList;
  5. import java.util.List;
  6. import java.util.StringTokenizer;
  8. import org.apache.hadoop.io.LongWritable;
  9. import org.apache.hadoop.io.Text;
  10. import org.apache.hadoop.mapreduce.Mapper;
  12. public class KnnMapper extends Mapper<LongWritable, Text, LongWritable, DistClassPair> {
  13.     private List<List<Object>> test;
  14.     private List<String> testY;
  16.     private static final double[] MIN = { 28, 1.58, 55, 22, -306, -271, -603, -494, -571, -616,
  17.             -499, -506, -613, -700, -213, -251 };
  18.     private static final double[] MAX = { 75, 1.71, 83, 28.6, 509, 533, 411, 69,
  19.             128, 102, 351, 471, -20, -13, -39, -56 };
  20.     // The map() method is run by MapReduce once for each row supplied as the
  21.     // input data
  22.     public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
  23.         /*
  24.          *  Tokenize the input line (presented as 'value' by MapReduce) from the csv file
  25.          */
  26.         String trainLine = value.toString();
  27.         StringTokenizer st = new StringTokenizer(trainLine, "\t");
  29.         List<Object> trainData = new ArrayList<>();
  31.         String user = st.nextToken();
  32.         String gender = st.nextToken();
  34.         trainData.add(user);
  35.         trainData.add(gender);
  37.         int index = 0;
  38.         while (st.countTokens() > 1) {
  39.             trainData.add(HelperFunc.normalize(st.nextToken(), MIN[index], MAX[index]));
  40.             index++;
  41.         }
  43.         String cls = st.nextToken();
  45.         /*
  46.          * Calculate the distance between each test data and train data,
  47.          * the variable index means the location of test data
  48.          */
  49.         index = 0;
  50.         for (List<Object> testData : test) {
  51.             double dist = HelperFunc.calcDistance(trainData, testData);
  52.             context.write(new LongWritable(index), new DistClassPair(dist, cls));
  53.             index++;
  54.         }
  55.     }
  57.     @Override
  58.     /*
  59.      * Set up the testing data from the cache
  60.      */
  61.     protected void setup(Context context) throws IOException, InterruptedException {
  62.         test = new ArrayList<>();
  63.         testY = new ArrayList<>();
  65.         BufferedReader buff = new BufferedReader(new FileReader(context.getCacheFiles()[0].toString()));
  66.         String line = buff.readLine();
  68.         System.out.println(line);
  70.         while (line != null) {
  71.             StringTokenizer st = new StringTokenizer(line, "\t");
  72.             List<Object> testData = new ArrayList<>();
  74.             String user = st.nextToken();
  75.             String gender = st.nextToken();
  77.             testData.add(user);
  78.             testData.add(gender);
  80.             int index = 0;
  81.             while (st.countTokens() > 1) {
  82.                 testData.add(HelperFunc.normalize(st.nextToken(), MIN[index], MAX[index]));
  83.                 index++;
  84.             }
  86.             test.add(testData);
  87.             testY.add(st.nextToken());
  88.             line = buff.readLine();
  89.         }
  91.         buff.close();
  92.     }
  94. }

在我的实现中，我使用了一个自定义类 DistClassPair ，它只是将距离和类信息作为值存储。

  1. import java.io.DataInput;
  2. import java.io.DataOutput;
  3. import java.io.IOException;
  5. import org.apache.hadoop.io.WritableComparable;
  7. public class DistClassPair implements WritableComparable<DistClassPair> {   
  8.     private Double dist;
  9.     private String cls;
  11.     public DistClassPair(Double dist, String cls) {
  12.         this.dist = dist;
  13.         this.cls = cls;
  14.     }
  16.     @Override
  17.     public void readFields(DataInput in) throws IOException {
  18.         dist = in.readDouble();
  19.         cls = in.readLine();
  20.     }
  22.     @Override
  23.     public void write(DataOutput out) throws IOException {
  24.         out.writeDouble(dist);
  25.         out.writeBytes(cls);
  26.     }
  28.     @Override
  29.     public int compareTo(DistClassPair o) {
  30.         return Double.compare(dist, o.dist);
  31.     }
  33.     public String getCls() {
  34.         return cls;
  35.     }
  36. }

下面是 KnnDriver 如果你愿意，我会写信的。

  1. import java.net.URI;
  2. import org.apache.hadoop.conf.Configuration;
  3. import org.apache.hadoop.conf.Configured;
  4. import org.apache.hadoop.fs.FileSystem;
  5. import org.apache.hadoop.fs.Path;
  6. import org.apache.hadoop.io.LongWritable;
  7. import org.apache.hadoop.io.Text;
  8. import org.apache.hadoop.mapreduce.Job;
  9. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  10. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
  11. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  12. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
  13. import org.apache.hadoop.util.Tool;
  14. import org.apache.hadoop.util.ToolRunner;
  16. public class KnnDriver extends Configured implements Tool{
  17.     /*
  18.      *  args = N, test.csv, train.csv, outputpath
  19.      */
  20.     public static void main(String[] args) throws Exception {
  21.         int res = ToolRunner.run(new Configuration(), new KnnDriver(), args);
  22.         System.exit(res);
  23.     }
  25.     @Override
  26.     public int run(String[] args) throws Exception {
  27.         Configuration conf = getConf();
  28.         conf.set("N", args[0]);
  30.         Job job = Job.getInstance(conf, "K-Nearest-Neighbor mapreduce");        
  31.         job.setJarByClass(KnnDriver.class);
  33.         job.addCacheFile(new URI(args[1]));
  35.         if (args.length != 4) {
  36.             System.err.println("Number of parameter is not correct!");
  37.             System.exit(2);
  38.         }
  40.         job.setMapperClass(KnnMapper.class);
  41.         job.setReducerClass(KnnReducer.class);
  43.         // TODO: specify output types
  44.         job.setOutputKeyClass(LongWritable.class);
  45.         job.setMapOutputValueClass(DistClassPair.class);
  46.         job.setOutputValueClass(Text.class);
  48.         job.setInputFormatClass(TextInputFormat.class);
  49.         job.setOutputFormatClass(TextOutputFormat.class);
  51.         // TODO: specify input and output DIRECTORIES (not files)
  52.         FileInputFormat.setInputPaths(job, new Path(args[2]));
  54.         Path outputPath = new Path(args[3]);
  55.         FileSystem.get(conf).delete(outputPath, true);
  56.         FileOutputFormat.setOutputPath(job, outputPath);
  58.         return(job.waitForCompletion(true) ? 0 : -1);
  59.     }
  61. }

非常感谢。

Javahadoopmapreducebigdataknn

暂无答案！

目前还没有任何答案，快来回答吧！

我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com