26 集群使用初步

[toto@hadoop software]$ hadoop fs -put ./findbugs-1.3.9 /

put: `/findbugs-1.3.9/LICENSE-ASM.txt': File exists

put: `/findbugs-1.3.9/LICENSE-AppleJavaExtensions.txt': File exists

put: `/findbugs-1.3.9/LICENSE-bcel.txt': File exists

put: `/findbugs-1.3.9/LICENSE-commons-lang.txt': File exists

put: `/findbugs-1.3.9/LICENSE-docbook.txt': File exists

put: `/findbugs-1.3.9/LICENSE-dom4j.txt': File exists

put: `/findbugs-1.3.9/LICENSE-jFormatString.txt': File exists

[toto@hadoop learn]$ cd /home/toto/learn

/home/toto/learn

[toto@hadoop learn]$ pwd

/home/toto/learn

[toto@hadoop learn]$ hadoop fs -get /findbugs-1.3.9/LICENSE-ASM.txt

[toto@hadoop learn]$ ls

LICENSE-ASM.txt

[toto@hadoop1 hadoop-2.8.0]$cd /home/toto/software/hadoop-2.8.0

[toto@hadoop1 hadoop-2.8.0]$sbin/start-dfs.sh

[toto@hadoop mapreduce]$ cd /home/toto/software/hadoop-2.8.0/share/hadoop/mapreduce

[toto@hadoop mapreduce]$ pwd

/home/toto/software/hadoop-2.8.0/share/hadoop/mapreduce

[toto@hadoop mapreduce]$ ll

总用量 5088

-rw-r--r--. 1 toto hadoop  562900 3月  17 13:31 hadoop-mapreduce-client-app-2.8.0.jar

-rw-r--r--. 1 toto hadoop  782739 3月  17 13:31 hadoop-mapreduce-client-common-2.8.0.jar

-rw-r--r--. 1 toto hadoop 1571179 3月  17 13:31 hadoop-mapreduce-client-core-2.8.0.jar

-rw-r--r--. 1 toto hadoop  195000 3月  17 13:31 hadoop-mapreduce-client-hs-2.8.0.jar

-rw-r--r--. 1 toto hadoop   31533 3月  17 13:31 hadoop-mapreduce-client-hs-plugins-2.8.0.jar

-rw-r--r--. 1 toto hadoop   66999 3月  17 13:31 hadoop-mapreduce-client-jobclient-2.8.0.jar

-rw-r--r--. 1 toto hadoop 1587158 3月  17 13:31 hadoop-mapreduce-client-jobclient-2.8.0-tests.jar

-rw-r--r--. 1 toto hadoop   75495 3月  17 13:31 hadoop-mapreduce-client-shuffle-2.8.0.jar

-rw-r--r--. 1 toto hadoop  301934 3月  17 13:31 hadoop-mapreduce-examples-2.8.0.jar

drwxr-xr-x. 2 toto hadoop    4096 3月  17 13:31 jdiff

drwxr-xr-x. 2 toto hadoop    4096 3月  17 13:31 lib

drwxr-xr-x. 2 toto hadoop    4096 3月  17 13:31 lib-examples

drwxr-xr-x. 2 toto hadoop    4096 3月  17 13:31 sources

[toto@hadoop mapreduce]$

使用命令运行mapreduce命令：

[toto@hadoop mapreduce]$ hadoop jar hadoop-mapreduce-examples-2.8.0.jar pi 5 5

Number of Maps  = 5

Samples per Map = 5

Wrote input for Map #0

Wrote input for Map #1

Wrote input for Map #2

Wrote input for Map #3

Wrote input for Map #4

Starting Job

17/05/29 14:47:36 INFO client.RMProxy: Connecting to ResourceManager at hadoop/192.168.106.80:8032

17/05/29 14:47:37 INFO input.FileInputFormat: Total input files to process : 5

17/05/29 14:47:37 INFO mapreduce.JobSubmitter: number of splits:5

17/05/29 14:47:38 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1495998405307_0001

17/05/29 14:47:39 INFO impl.YarnClientImpl: Submitted application application_1495998405307_0001

17/05/29 14:47:39 INFO mapreduce.Job: The url to track the job: http://hadoop:8088/proxy/application_1495998405307_0001/

17/05/29 14:47:39 INFO mapreduce.Job: Running job: job_1495998405307_0001

17/05/29 14:48:00 INFO mapreduce.Job: Job job_1495998405307_0001 running in uber mode : false

17/05/29 14:48:00 INFO mapreduce.Job:  map 0% reduce 0%

//首先要定义四个泛型的类型

//keyin:  LongWritable    valuein: Text

//keyout: Text            valueout:IntWritable

public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{

         //map方法的生命周期：  框架每传一行数据就被调用一次

         //key :  这一行的起始点在文件中的偏移量

         //value: 这一行的内容

         @Override

         protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

                   //拿到一行数据转换为string

                   String line = value.toString();

                   //将这一行切分出各个单词

                   String[] words = line.split(" ");

                   //遍历数组，输出<单词，1>

                   for(String word:words){

                            context.write(new Text(word), new IntWritable(1));

                   }

         }

}

         //生命周期：框架每传递进来一个kv 组，reduce方法被调用一次

         @Override

         protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {

                   //定义一个计数器

                   int count = 0;

                   //遍历这一组kv的所有v，累加到count中

                   for(IntWritable value:values){

                            count += value.get();

                   }

                   context.write(key, new IntWritable(count));

         }

}

public class WordCountRunner {

         //把业务逻辑相关的信息（哪个是mapper，哪个是reducer，要处理的数据在哪里，输出的结果放哪里。。。。。。）描述成一个job对象

         //把这个描述好的job提交给集群去运行

         public static void main(String[] args) throws Exception {

                   Configuration conf = new Configuration();

                   Job wcjob = Job.getInstance(conf);

                   //指定我这个job所在的jar包

//               wcjob.setJar("/home/hadoop/wordcount.jar");

                   wcjob.setJarByClass(WordCountRunner.class);

                   wcjob.setMapperClass(WordCountMapper.class);

                   wcjob.setReducerClass(WordCountReducer.class);

                   //设置我们的业务逻辑Mapper类的输出key和value的数据类型

                   wcjob.setMapOutputKeyClass(Text.class);

                   wcjob.setMapOutputValueClass(IntWritable.class);

                   //设置我们的业务逻辑Reducer类的输出key和value的数据类型

                   wcjob.setOutputKeyClass(Text.class);

                   wcjob.setOutputValueClass(IntWritable.class);

                   //指定要处理的数据所在的位置

                   FileInputFormat.setInputPaths(wcjob, "hdfs://hdp-server01:9000/wordcount/data/big.txt");

                   //指定处理完成之后的结果所保存的位置

                   FileOutputFormat.setOutputPath(wcjob, new Path("hdfs://hdp-server01:9000/wordcount/output/"));

                   //向yarn集群提交这个job

                   boolean res = wcjob.waitForCompletion(true);

                   System.exit(res?0:1);

         }

Hello tom

Hello jim

Hello ketty

Hello world

Ketty tom