hadoop数据集mapreduce datajoin

mfuanj7w  于 2021-06-03  发布在  Hadoop
关注(0)|答案(0)|浏览(268)

代码
我试着在ActionBook中运行hadoop的datajoin示例。

  1. import java.io.DataInput;
  2. import java.io.DataOutput;
  3. import java.io.IOException;
  4. // import org.apache.commons.logging.Log;
  5. // import org.apache.commons.logging.LogFactory;
  6. import org.apache.hadoop.mapred.FileInputFormat;
  7. import org.apache.hadoop.mapred.FileOutputFormat;
  8. import org.apache.hadoop.mapred.JobClient;
  9. import org.apache.hadoop.mapred.JobConf;
  10. import org.apache.hadoop.mapred.TextInputFormat;
  11. import org.apache.hadoop.mapred.TextOutputFormat;
  12. import org.apache.hadoop.util.GenericOptionsParser;
  13. import org.apache.hadoop.util.Tool;
  14. import org.apache.hadoop.util.ToolRunner;
  15. import org.apache.hadoop.fs.Path;
  16. import org.apache.hadoop.io.Text;
  17. import org.apache.hadoop.io.Writable;
  18. import org.apache.hadoop.conf.Configuration;
  19. import org.apache.hadoop.conf.Configured;
  20. import org.apache.hadoop.contrib.utils.join.*;
  22. public class MultiDataSetJoinMR extends Configured implements Tool
  23. {
  24.     public static class MapClass extends DataJoinMapperBase
  25.     {
  26.         protected Text generateInputTag(String inputFile)
  27.         {
  28.             String datasource = inputFile.split("-")[0];
  29.             return new Text(datasource);
  30.         }
  31.         protected Text generateGroupKey(TaggedMapOutput aRecord)
  32.         {
  33.             String line = ((Text) aRecord.getData()).toString();
  34.             String[] tokens = line.split(",");
  35.             String groupKey = tokens[0];
  36.             return new Text(groupKey);
  37.         }
  38.         protected TaggedMapOutput generateTaggedMapOutput(Object value)
  39.         {
  40.             TaggedWritable retv = new TaggedWritable((Text) value);
  41.             retv.setTag(this.inputTag);
  42.             return retv;
  43.         }
  44.     }
  45.     public static class Reduce extends DataJoinReducerBase
  46.     {
  47.         protected TaggedMapOutput combine(Object[] tags, Object[] values)
  48.         {
  49.             if (tags.length < 2) return null;
  50.             String joinedStr = "";
  51.             for (int i=0; i<values.length; i++)
  52.             {
  53.                 if (i > 0) joinedStr += ",";
  54.                 TaggedWritable tw = (TaggedWritable) values[i];
  55.                 String line = ((Text) tw.getData()).toString();
  56.                 String[] tokens = line.split(",", 2);
  57.                 joinedStr += tokens[1];
  58.             }
  59.             TaggedWritable retv = new TaggedWritable(new Text(joinedStr));
  60.             retv.setTag((Text) tags[0]);
  61.             return retv;
  62.         }
  63.     }
  64.     public static class TaggedWritable extends TaggedMapOutput
  65.     {
  66.         private Writable data;
  68.         public TaggedWritable() {
  69.             this.tag = new Text();
  70.         }
  72.         public TaggedWritable(Writable data)
  73.         {
  74.             this.tag = new Text("");
  75.             this.data = data;
  76.         }
  78.         public Writable getData()
  79.         {
  80.             return data;
  81.         }
  82.         public void write(DataOutput out) throws IOException
  83.         {
  84.             this.tag.write(out);
  85.             this.data.write(out);
  86.         }
  87.         public void readFields(DataInput in) throws IOException
  88.         {
  89.             this.tag.readFields(in);
  90.             this.data.readFields(in);
  91.         }
  92.     }
  93.     public int run(String[] args) throws Exception
  94.     {
  96.         Configuration conf = getConf();
  97.         JobConf job = new JobConf(conf, MultiDataSetJoinMR.class);
  98.         String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
  99.         if (otherArgs.length != 2)
  100.         {
  101.             System.err.println("Usage: wordcount <in> <out>");
  102.             System.exit(2);
  103.         }
  105.         Path in = new Path(args[0]);
  106.         Path out = new Path(args[1]);
  107.         FileInputFormat.setInputPaths(job, in);
  108.         FileOutputFormat.setOutputPath(job, out);
  109.         job.setJobName("DataJoin");
  110.         job.setMapperClass(MapClass.class);
  111.         job.setReducerClass(Reduce.class);
  112.         job.setInputFormat(TextInputFormat.class);
  113.         job.setOutputFormat(TextOutputFormat.class);
  114.         job.setOutputKeyClass(Text.class);
  115.         job.setOutputValueClass(TaggedWritable.class);
  116.         job.set("mapred.textoutputformat.separator", ",");
  117.         JobClient.runJob(job);
  118.         return 0;
  119.     }
  120.     public static void main(String[] args) throws Exception
  121.     {
  122.         int res = ToolRunner.run(new Configuration(),
  123.                 new MultiDataSetJoinMR(),
  124.                 args);
  125.         System.exit(res);
  126.     }
  127. }

运行命令
./hadoop jar multidatasetjoin.jar/home/project/dataset/home/project/out
错误
但我面临以下问题。

  1. 15 Mar, 2013 4:29:45 PM org.apache.hadoop.metrics.jvm.JvmMetrics init
  2. INFO: Initializing JVM Metrics with processName=JobTracker, sessionId=
  3. 15 Mar, 2013 4:29:45 PM org.apache.hadoop.mapred.JobClient configureCommandLineOptions
  4. WARNING: No job jar file set.  User classes may not be found. See JobConf(Class) or JobConf#setJar(String).
  5. 15 Mar, 2013 4:29:45 PM org.apache.hadoop.mapred.FileInputFormat listStatus
  6. INFO: Total input paths to process : 2
  7. 15 Mar, 2013 4:29:45 PM org.apache.hadoop.mapred.JobClient monitorAndPrintJob
  8. INFO: Running job: job_local_0001
  9. 15 Mar, 2013 4:29:45 PM org.apache.hadoop.mapred.FileInputFormat listStatus
  10. INFO: Total input paths to process : 2
  11. 15 Mar, 2013 4:29:45 PM org.apache.hadoop.mapred.MapTask runOldMapper
  12. INFO: numReduceTasks: 1
  13. 15 Mar, 2013 4:29:45 PM org.apache.hadoop.mapred.MapTask$MapOutputBuffer <init>
  14. INFO: io.sort.mb = 100
  15. 15 Mar, 2013 4:29:45 PM org.apache.hadoop.mapred.MapTask$MapOutputBuffer <init>
  16. INFO: data buffer = 79691776/99614720
  17. 15 Mar, 2013 4:29:45 PM org.apache.hadoop.mapred.MapTask$MapOutputBuffer <init>
  18. INFO: record buffer = 262144/327680
  19. 15 Mar, 2013 4:29:45 PM org.apache.hadoop.mapred.LocalJobRunner$Job run
  20. WARNING: job_local_0001
  21. java.lang.RuntimeException: Error in configuring object
  22.     at org.apache.hadoop.util.ReflectionUtils.setJobConf(ReflectionUtils.java:93)
  23.     at org.apache.hadoop.util.ReflectionUtils.setConf(ReflectionUtils.java:64)
  24.     at org.apache.hadoop.util.ReflectionUtils.newInstance(ReflectionUtils.java:117)
  25.     at org.apache.hadoop.mapred.MapTask.runOldMapper(MapTask.java:354)
  26.     at org.apache.hadoop.mapred.MapTask.run(MapTask.java:307)
  27.     at org.apache.hadoop.mapred.LocalJobRunner$Job.run(LocalJobRunner.java:177)
  28. Caused by: java.lang.reflect.InvocationTargetException
  29.     at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
  30.     at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
  31.     at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
  32.     at java.lang.reflect.Method.invoke(Method.java:616)
  33.     at org.apache.hadoop.util.ReflectionUtils.setJobConf(ReflectionUtils.java:88)
  34.     ... 5 more
  35. Caused by: java.lang.RuntimeException: Error in configuring object
  36.     at org.apache.hadoop.util.ReflectionUtils.setJobConf(ReflectionUtils.java:93)
  37.     at org.apache.hadoop.util.ReflectionUtils.setConf(ReflectionUtils.java:64)
  38.     at org.apache.hadoop.util.ReflectionUtils.newInstance(ReflectionUtils.java:117)
  39.     at org.apache.hadoop.mapred.MapRunner.configure(MapRunner.java:34)
  40.     ... 10 more
  41. Caused by: java.lang.reflect.InvocationTargetException
  42.     at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
  43.     at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
  44.     at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
  45.     at java.lang.reflect.Method.invoke(Method.java:616)
  46.     at org.apache.hadoop.util.ReflectionUtils.setJobConf(ReflectionUtils.java:88)
  47.     ... 13 more
  48. Caused by: java.lang.NullPointerException
  49.     at MultiDataSetJoinMR$MapClass.generateInputTag(MultiDataSetJoinMR.java:31)
  50.     at org.apache.hadoop.contrib.utils.join.DataJoinMapperBase.configure(DataJoinMapperBase.java:60)
  51.     ... 18 more
  52. null15 Mar, 2013 4:29:46 PM org.apache.hadoop.mapred.JobClient monitorAndPrintJob
  53. INFO:  map 0% reduce 0%
  54. 15 Mar, 2013 4:29:46 PM org.apache.hadoop.mapred.JobClient monitorAndPrintJob
  55. INFO: Job complete: job_local_0001
  56. 15 Mar, 2013 4:29:46 PM org.apache.hadoop.mapred.Counters log
  57. INFO: Counters: 0
  58. Exception in thread "main" java.io.IOException: Job failed!
  59.     at org.apache.hadoop.mapred.JobClient.runJob(JobClient.java:1252)
  60.     at MultiDataSetJoinMR.run(MultiDataSetJoinMR.java:123)
  61.     at org.apache.hadoop.util.ToolRunner.run(ToolRunner.java:65)
  62.     at MultiDataSetJoinMR.main(MultiDataSetJoinMR.java:128)
  63.     at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
  64.     at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
  65.     at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
  66.     at java.lang.reflect.Method.invoke(Method.java:616)
  67.     at com.intellij.rt.execution.application.AppMain.main(AppMain.java:120)

从日志跟踪中，我可以确定inputfile变量在下面的方法中获得空值，

  1. protected Text generateInputTag(String inputFile)
  2. {
  3.     String datasource = inputFile.split("-")[0];
  4.     return new Text(datasource);
  5. }

我不知道从哪里来的电话，以及如何解决它。有人能帮我吗

hadoopmapreduceinner-join

暂无答案！

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

我来回答

相关问题

    查看更多

    热门标签

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

    最新问答

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