hadoop记录读取器只读取第一行，然后输入流似乎被关闭

8dtrkrch  于 2021-06-04  发布在  Hadoop
关注(0)|答案(0)|浏览(317)

我正在尝试实现一个hadoop作业，它统计对象（单击）在数据集中出现的频率。因此，我编写了一个自定义文件输入格式。记录读取器似乎只读取给定文件的第一行，并关闭输入流。
代码如下：
pojo类：

  1. package model;
  2. import java.io.DataInput;
  3. import java.io.DataOutput;
  4. import java.io.IOException;
  5. import org.apache.hadoop.io.WritableComparable;
  7. public class Click implements WritableComparable<Click>  {
  8.     private String user;
  9.     private String clickStart;
  10.     private String date;
  11.     private String clickTarget;
  13.     @Override
  14.     public void write(DataOutput out) throws IOException {
  15.         out.writeUTF(user);
  16.         out.writeUTF(clickStart);
  17.         out.writeUTF(date);
  18.         out.writeUTF(clickTarget);
  19.     }
  20.     @Override
  21.     public void readFields(DataInput in) throws IOException {
  22.         user = in.readUTF();
  23.         clickStart = in.readUTF();
  24.         date = in.readUTF();
  25.         clickTarget = in.readUTF();
  26.     }
  28.     public int compareTo(Click arg0) {
  29.         int response = clickTarget.compareTo(arg0.clickTarget);
  30.             if (response == 0) {
  31.                 response = date.compareTo(arg0.date);
  32.             }
  33.         return response;
  34.     }
  35.     public String getUser(String user) {
  36.         return this.user;
  37.     }
  38.     public void setUser(String user) {
  39.         this.user = user;
  40.     }
  42.     public String getClickStart() {
  43.         return clickStart;
  44.     }
  46.     public void setClickStart(String clickStart) {
  47.         this.clickStart = clickStart;
  48.     }
  50.     public String getDate() {
  51.         return date;
  52.     }
  54.     public void setDate(String date) {
  55.         this.date = date;
  56.     }
  58.     public String getClickTarget() {
  59.         return clickTarget;
  60.     }
  62.     public void setClickTarget(String clickTarget) {
  63.         this.clickTarget = clickTarget;
  64.     }
  66.     public String toString() {
  67.         return clickStart + "\t" + date;
  68.     }
  69. }

以下是fileinputformat类：

  1. package ClickAnalysis;
  3. import java.io.BufferedReader;
  4. import java.io.IOException;
  5. import java.io.InputStream;
  6. import java.io.InputStreamReader;
  8. import model.Click;
  10. import org.apache.hadoop.conf.Configuration;
  11. import org.apache.hadoop.fs.Path;
  12. import org.apache.hadoop.io.IntWritable;
  13. import org.apache.hadoop.mapreduce.lib.input.FileSplit;
  14. import org.apache.hadoop.mapreduce.InputSplit;
  15. import org.apache.hadoop.mapreduce.RecordReader;
  16. import org.apache.hadoop.mapreduce.TaskAttemptContext;
  17. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  18. import org.apache.hadoop.util.StringUtils;
  19. import org.apache.tools.ant.types.CommandlineJava.SysProperties;
  21. public class ClickAnalysisInputFormat extends FileInputFormat<Click, IntWritable>{
  23.     @Override
  24.     public RecordReader<Click, IntWritable> createRecordReader(
  25.             InputSplit split, TaskAttemptContext context) throws IOException,
  26.             InterruptedException {
  27.         System.out.println("Creating Record Reader");
  28.         return new ClickReader();
  29.     }
  31.     public static class ClickReader extends RecordReader<Click, IntWritable> {
  32.         private BufferedReader in;
  33.         private Click key;
  34.         private IntWritable value;
  36.         @Override
  37.         public void initialize(InputSplit inputSplit, TaskAttemptContext context) throws IOException, InterruptedException {
  38.             key = new Click();
  39.             value = new IntWritable(1);
  40.             System.out.println("Starting to read ...");
  42.             FileSplit split = (FileSplit) inputSplit;
  43.             Configuration conf = context.getConfiguration();
  45.             Path path = split.getPath();
  46.             InputStream is = path.getFileSystem(conf).open(path);
  47.             in = new BufferedReader(new InputStreamReader(is));
  48.         }
  50.         @Override
  51.         public boolean nextKeyValue() throws IOException, InterruptedException {
  52.             String line = in.readLine();
  53.             System.out.println("line: " + line);
  54.             boolean hasNextKeyValue;
  55.             if (line == null) {
  56.                 System.out.println("line is null");
  57.                 hasNextKeyValue = false;
  58.             } else {
  59.                 String[] click = StringUtils.split(line, '\\', ';');
  60.                 System.out.println(click[0].toString());
  61.                 System.out.println(click[1].toString());
  62.                 System.out.println(click[2].toString());
  64.                 key.setClickStart(click[0].toString());
  65.                 key.setDate(click[1].toString());
  66.                 key.setClickTarget(click[2].toString());
  67.                 value.set(1);
  68.                 System.out.println("done with first line");
  69.                 hasNextKeyValue = true;
  70.             }
  71.             System.out.println(hasNextKeyValue);
  72.             return hasNextKeyValue;
  73.         }
  75.         @Override
  76.         public Click getCurrentKey() throws IOException, InterruptedException {
  77.             return this.key;
  78.         }
  80.         @Override
  81.         public IntWritable getCurrentValue() throws IOException, InterruptedException {
  82.             return this.value;
  83.         }
  85.         @Override
  86.         public float getProgress() throws IOException, InterruptedException {
  87.             return 0;
  88.         }
  90.         public void close() throws IOException {
  91.             in.close();
  92.             System.out.println("in closed");
  93.         }
  94.     }
  95. }

Map器类：

  1. package ClickAnalysis;
  3. import java.io.IOException;
  5. import model.Click;
  6. import model.ClickStartTarget;
  8. import org.apache.hadoop.io.*;
  9. import org.apache.hadoop.mapreduce.Mapper;
  10. import org.jruby.RubyProcess.Sys;
  12. public class ClickAnalysisMapper extends Mapper<Click, IntWritable, Click, IntWritable> {
  14.     private static final IntWritable outputValue = new IntWritable();
  16.     @Override
  17.     protected void map(Click key, IntWritable value, Context context) throws IOException, InterruptedException {
  18.         System.out.println("Key: " + key.getClickStart() + " " + key.getDate() + " " + key.getClickTarget() + " Value: " + value);
  20.         outputValue.set(value.get());
  21.         System.out.println(outputValue.get());
  22.         context.write(key, outputValue);
  23.         System.out.println("nach context");
  24.     }
  25. }

分区器类：

  1. package ClickAnalysis;
  3. import model.Click;
  5. import org.apache.hadoop.io.IntWritable;
  6. import org.apache.hadoop.mapreduce.Partitioner;
  8. public class ClickAnalysisPartitioner extends Partitioner<Click, IntWritable> {
  10.     @Override
  11.     public int getPartition(Click key, IntWritable value, int numPartitions) {
  12.         System.out.println("in Partitioner drinnen");
  13.         int partition = numPartitions;
  14.         return partition;
  15.     }
  16. }

hadoop作业，通过servlet容器中的restful web服务调用触发，但这不应该是问题所在：

  1. package ClickAnalysis;
  3. import model.Click;
  5. import org.apache.hadoop.conf.Configuration;
  6. import org.apache.hadoop.fs.Path;
  7. import org.apache.hadoop.io.IntWritable;
  8. import org.apache.hadoop.mapreduce.Job;
  9. import org.apache.hadoop.mapreduce.*;
  10. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  11. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  12. import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
  14. public class ClickAnalysisJob {
  16.     public int run() throws Exception {
  18.         // TODO Auto-generated method stub
  19.         Configuration conf = new Configuration();
  21.         Job job = Job.getInstance(conf, "ClickAnalysisJob");
  22.         job.setJarByClass(ClickAnalysisJob.class);
  24.         // Job Input path
  25.         FileInputFormat.setInputPaths(job, "hdfs://localhost:9000/user/hadoop/testdata1.csv");
  26.         // Job Output path
  28.         Path out = new Path("hdfs://localhost:9000/user/hadoop/clickdataAnalysis_out");
  29.         FileOutputFormat.setOutputPath(job, out);
  30.         out.getFileSystem(conf).delete(out,true);
  32.         job.setMapperClass(ClickAnalysisMapper.class);
  33.         job.setReducerClass(Reducer.class);
  34.         job.setPartitionerClass(ClickAnalysisPartitioner.class);
  35.         //job.setReducerClass(ClickAnalysisReducer.class);
  37.         job.setInputFormatClass(ClickAnalysisInputFormat.class);
  38.         job.setOutputFormatClass(SequenceFileOutputFormat.class);
  39.         job.setOutputKeyClass(Click.class);
  40.         job.setOutputValueClass(IntWritable.class);
  41.         job.setMapOutputKeyClass(Click.class);
  42.         job.setMapOutputValueClass(IntWritable.class);
  44.         System.out.println("in run drinnen");
  46.         //job.setGroupingComparatorClass(ClickTargetAnalysisComparator.class);
  48.         job.setNumReduceTasks(1);
  50.         int result = job.waitForCompletion(true)? 0:1;
  51.         return result;
  52.     }
  53. }

接下来是数据集（示例）：

  1. /web/big-data-test-site/test-seite-1;2014-07-08;ein ziel
  2. /web/big-data-test-site/test-seite-1;2014-07-08;ein anderes ziel
  3. /web/big-data-test-site/test-seite-1;2014-07-08;ein anderes ziel
  4. /web/big-data-test-site/test-seite-1;2014-07-08;ein ziel
  5. /web/big-data-test-site/test-seite-1;2014-07-08;ein drittes ziel
  6. /web/big-data-test-site/test-seite-1;2014-07-08;ein ziel
  7. /web/big-data-test-site/test-seite-1;2014-07-08;ein viertes ziel
  8. /web/big-data-test-site/test-seite-1;2014-07-08;ein ziel

当我运行程序时，syso显示如下：

  1. in run drinnen
  2. Creating Record Reader
  3. Starting to read ...
  4. line: /web/big-data-test-site/test-seite-1;2014-07-08;ein ziel
  5. /web/big-data-test-site/test-seite-1
  6. 2014-07-08
  7. ein ziel
  8. done with first line
  9. true
  10. Key: /web/big-data-test-site/test-seite-1 2014-07-08 ein ziel Value: 1
  11. 1
  12. in closed
  13. analyze Method: 1

由此我得出结论，读录器只读第一行。为什么会发生这种情况，如何解决？

hadoopapachehadoop2recordreader

暂无答案！

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

我来回答

相关问题

    查看更多

    热门标签

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

    最新问答

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