Hadoop-2.2.0 (传 hadoop-2.2.0.tar.gz)彩民之家论坛9

2019-10-11 05:36 来源:未知

Python-MapReduce

编写mapper代码,map.py

import sysfor line in sys.stdin: fields = line.strip for item in fields: print(item   ' '   '1')

编写reducer代码,reduce.py

import sysresult = {}for line in sys.stdin: kvs = line.strip().split k = kvs[0] v = kvs[1] if k in result: result[k] =1 else: result[k] = 1for k,v in result.items(): print("%st%s" %

加上测量试验文本,test1.txt

tale as old as timetrue as it can bebeauty and the beast

cat test1.txt | python map.py结果:

tale 1as 1old 1as 1time 1true 1as 1it 1can 1be 1beauty 1and 1the 1beast 1

cat test1.txt | python map.py | sort -k1,1 | python reduce.py

进行结果:

and 1be 1old 1beauty 1true 1it 1beast 1as 3can 1time 1the 1tale 1

本土测量试验截止,编写脚本在HDFS中实施顺序

脚本:run.sh (请依据本机情况修改)

HADOOP_CMD="/app/hadoop-3.1.2/bin/hadoop"STREAM_JAR_PATH="/app/hadoop-3.1.2/share/hadoop/tools/lib/hadoop-streaming-3.1.2.jar"INPUT_FILE_PATH_1="/py/input/"OUTPUT_PATH="/output"$HADOOP_CMD fs -rmr-skipTrash $OUTPUT_PATH# Step 1.$HADOOP_CMD jar $STREAM_JAR_PATH -input $INPUT_FILE_PATH_1 -output $OUTPUT_PATH -mapper "python map.py" -reducer "python reduce.py" -file ./map.py -file ./reduce.py 

丰盛进行权限chmod a x run.sh;实施测验:bash run.sh,查看结果:

彩民之家论坛9066777 1image.png

二,编写程序达成对输入文件的排序:

  1. ##### 未来有多少个输入文件,每一个文件中的每行内容均为八个大背头。须求读取全部文件中的整数,实行升序排序后,输出到三个新的公文中,输出的数量格式为每行多少个整数,第二个数字为第四个整数的排序位次,第一个整数为原待排列的卡尺头。下边是输入文件和输出文件的一个样例供参谋。
  • 输入文件file1.txt的样举例下:
33
37
12
40
  • 输入文件file2.txt的样举例下:
4
16
39
5
  • 输入文件file3.txt的样举个例子下:
1
45
25
  • 依靠输入文件file1.txt、file2.txt和file3.txt获得的输出文件如下:
1 1
2 4
3 5
4 12
5 16
6 25
7 33
8 37
9 39
10 40
11 45

试验进程:

  1. 创造文件file1.txt、file2.txt和file3.txt
将上面样例内容复制进去
  1. 在HDFS建立input2文件夹
  1. 上传样例到HDFS中的input2文件夹
  1. 到eclipse上实行代码

试验代码:

package cn.edu.zucc.mapreduce;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class ContentSort {

    public static class Map extends Mapper<Object, Text, IntWritable, IntWritable> {
        private static IntWritable data = new IntWritable();

        @Override
        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            data.set(Integer.parseInt(line));
            context.write(data, new IntWritable(1));
        }
    }

    public static class Reduce extends Reducer<IntWritable, IntWritable, IntWritable, IntWritable> {
        private static IntWritable linenum = new IntWritable(1);

        @Override
        public void reduce(IntWritable key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            for (IntWritable val : values) {
                context.write(linenum, key);
                linenum = new IntWritable(linenum.get()   1);
            }

        }

    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "hdfs://localhost:9000");
        String[] otherArgs = new String[]{"input2", "output2"};
        if (otherArgs.length != 2) {
            System.err.println("Usage: ContentSort <in> <out>");
            System.exit(2);
        }
        Job job = Job.getInstance(conf, "ContentSort");
        job.setJarByClass(ContentSort.class);
        job.setMapperClass(Map.class);
        job.setReducerClass(Reduce.class);
        job.setOutputKeyClass(IntWritable.class);
        job.setOutputValueClass(IntWritable.class);
        FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
        FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
        System.exit(job.waitForCompletion(true) ? 0 : 1);

    }

}

HDFS 的shell

使用Python操作HDFS

第一必要安装和导入hdfs库,使用pip install hdfs

from hdfs import * client = Client('http://ip:port') #2.X版本port 使用50070 3.x版本port 使用9870client.list #查看hdfs /下的目录

client.makedirsclient.makedirs('/test',permision = 777 ) # permision可以设置参数

client.rename('/test','123') #将/test 目录改名为123client.delete('/test',True) #第二个参数表示递归删除 

/test/log.txt 文件下载至/home目录下。

client.download('/test/log.txt','/home') 

with client.read("/test/[PPT]Google Protocol Buffers.pdf") as reader: print reader.read()

另外参数:

  • read(*args, **kwds)
  • hdfs_path:hdfs路径
  • offset:设置开端的字节地点
  • l- ength:读取的尺寸
  • buffer_size:用于传输数据的字节的缓冲区的分寸。暗中同意值设置在HDFS配置。
  • encoding:指确定人员编制码
  • chunk_size:假若设置为正数,上下文管理器将赶回三个产生器发生的每一chunk_size字节并非贰个好像文件的对象
  • delimiter:假诺设置,上下文管理器将回来三个产生器爆发每一趟碰着分隔符。此参数要求钦点的编码。
  • progress:回调函数来追踪进度,为每一chunk_size字节(不可用,借使块大小不是钦点)。它将传递四个参数,文件上传的门路和传导的字节数。称为贰次与- 1用作第二个参数。

将文件上传至hdfs的 /test下。

client.upload(‘/test’,’/home/test/a.log’)
  • 操作系统:Ubuntu-16.04
  • Hadoop版本:2.6.0
  • JDK版本:1.8
  • IDE:Eclipse

使用 more authorized_keys 查看

环境

条件使用:hadoop3.1,Python3.6,ubuntu18.04

Hadoop是接纳Java开拓的,推荐应用Java操作HDFS。

突发性也急需大家选取Python操作HDFS。

此次大家来谈谈哪些利用Python操作HDFS,实行理文件件上传,下载,查看文件夹,以至怎么样行使Python举办MapReduce编制程序。

一,编程完毕文件合併和去重操作:

  1. ##### 对于五个输入文件,即文件A和文件B,请编写MapReduce程序,对三个公文进行联合,并删除中间重复的内容,得到二个新的输出文件C。上面是输入文件和输出文件的一个样例供参谋。
  • 输入文件f1.txt的样举个例子下:
20150101     x
20150102     y
20150103     x
20150104     y
20150105     z
20150106     x
  • 输入文件f2.txt的样举个例子下:
20150101     y
20150102     y
20150103     x
20150104     z
20150105     y
  • 依据输入文件f1和f2合併获得的出口文件的样举例下:
20150101      x
20150101      y
20150102      y
20150103      x
20150104      y
20150104      z
20150105      y
20150105      z
20150106      x

试验进程:

  1. 创设文件f1.txt和f2.txt
将上面样例内容复制进去
  1. 在HDFS建构input文件夹(实行那步事先要敞开hadoop相关进度)
  1. 上传样例到HDFS中的input文件夹
  1. 继之张开eclipse
    Eclipse的使用
    1. 点开项目,找到 src 文件夹,右键选取 New -> Class
2.  输入 Package 和 Name,然后Finish



3.  写好Java代码(给的代码里要修改HDFS和本地路径),右键选择 Run As
    -&gt; Run on Hadoop,结果在HDFS系统中查看

尝试代码:

package cn.edu.zucc.mapreduce;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class Merge {

    public static class Map extends Mapper<Object, Text, Text, Text> {
        private static Text text = new Text();

        @Override
        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            text = value;
            context.write(text, new Text(""));
        }
    }

    public static class Reduce extends Reducer<Text, Text, Text, Text> {
        @Override
        public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            context.write(key, new Text(""));
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "hdfs://localhost:9000");
        String[] otherArgs = new String[]{"input", "output"};
        if (otherArgs.length != 2) {
            System.err.println("Usage: Merge and duplicate removal <in> <out>");
            System.exit(2);
        }
        Job job = Job.getInstance(conf, "Merge");
        job.setJarByClass(Merge.class);
        job.setMapperClass(Map.class);
        job.setReducerClass(Reduce.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
        FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }

}
scp id_dsa.pub root@192.168.1.202:~

练习

输入文件file1的样比如下:20150101 x``20150102 y``20150103 x``20150104 y``20150105 z``20150106 x

输入文件file2的样比如下:20150101 y``20150102 y``20150103 x``20150104 z``20150105 y

基于输入文件file1file2合併获得的输出文件file3的样比方下:

20150101 x``20150101 y``20150102 y``20150103 x``20150104 y``20150104 z``20150105 y``20150105 z``20150106 x

对此八个输入文件,即文件file1和文件file2,请编写MapReduce程序,对七个公文进行联合,并删除中间重复的内容,得到三个新的出口文件file3。为了完成文件合併去重的任务,你编写的主次要能将包涵重复内容的例外文件合併到二个尚无重新的重组文件,法规如下:

  • 第一列按学号排列;
  • 学号一样,按x,y,z排列。

输入文件内容如下:child parent``Steven Lucy``Steven Jack``Jone Lucy``Jone Jack``Lucy Mary``Lucy Frank``Jack Alice``Jack Jesse``David Alice``David Jesse``Philip David``Philip Alma``Mark David``Mark Alma

出口文件内容如下:

grandchild grandparent``Steven Alice``Steven Jesse``Jone Alice``Jone Jesse``Steven Mary``Steven Frank``Jone Mary``Jone Frank``Philip Alice``Philip Jesse``Mark Alice``Mark Jesse

你编写的次序要能发掘父亲和儿子辈关系,给出祖孙辈关系的报表。准绳如下:

  • 孙子在前,祖父在后
  • 儿子同样,祖父的名字依照A-Z排列

尝试内容和须要

第三个:hdfs-site.xml

实验目标

  文件包罗:

三,对给定的报表举办音讯开采:

  1. ##### 上边给出一个child-parent的报表,须要挖掘内部的父亲和儿子辈关系,给出祖孙辈关系的表格。
  • 输入文件table.txt内容如下:
child parent
Steven Lucy
Steven Jack
Jone Lucy
Jone Jack
Lucy Mary
Lucy Frank
Jack Alice
Jack Jesse
David Alice
David Jesse
Philip David
Philip Alma
Mark David
Mark Alma
  • 出口文件内容如下:
grandchild  grandparent
Mark    Jesse
Mark    Alice
Philip  Jesse
Philip  Alice
Jone    Jesse
Jone    Alice
Steven  Jesse
Steven  Alice
Steven  Frank
Steven  Mary
Jone    Frank
Jone    Mary

实验进程:

  1. 创设文件table
将上面样例内容复制进去
  1. 在HDFS建立input3文件夹
  1. 上传样例到HDFS中的input3文件夹
  1. 到eclipse上实施代码

实验代码:

package cn.edu.zucc.mapreduce;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class STJoin {
    public static int time = 0;

    public static class Map extends Mapper<Object, Text, Text, Text> {
        @Override
        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            String[] childAndParent = line.split(" ");
            List<String> list = new ArrayList<>(2);
            for (String childOrParent : childAndParent) {
                if (!"".equals(childOrParent)) {
                    list.add(childOrParent);
                }
            }
            if (!"child".equals(list.get(0))) {
                String childName = list.get(0);
                String parentName = list.get(1);
                String relationType = "1";
                context.write(new Text(parentName), new Text(relationType   " "
                          childName   " "   parentName));
                relationType = "2";
                context.write(new Text(childName), new Text(relationType   " "
                          childName   " "   parentName));
            }
        }
    }

    public static class Reduce extends Reducer<Text, Text, Text, Text> {
        @Override
        public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            if (time == 0) {
                context.write(new Text("grand_child"), new Text("grand_parent"));
                time  ;
            }
            List<String> grandChild = new ArrayList<>();
            List<String> grandParent = new ArrayList<>();
            for (Text text : values) {
                String s = text.toString();
                String[] relation = s.split("\ ");
                String relationType = relation[0];
                String childName = relation[1];
                String parentName = relation[2];
                if ("1".equals(relationType)) {
                    grandChild.add(childName);
                } else {
                    grandParent.add(parentName);
                }
            }
            int grandParentNum = grandParent.size();
            int grandChildNum = grandChild.size();
            if (grandParentNum != 0 && grandChildNum != 0) {
                for (int m = 0; m < grandChildNum; m  ) {
                    for (int n = 0; n < grandParentNum; n  ) {
                        context.write(new Text(grandChild.get(m)), new Text(
                                grandParent.get(n)));
                    }
                }
            }
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "hdfs://localhost:9000");
        String[] otherArgs = new String[]{"input3", "output3"};
        if (otherArgs.length != 2) {
            System.err.println("Usage: Single Table Join <in> <out>");
            System.exit(2);
        }
        Job job = Job.getInstance(conf, "Single table Join ");
        job.setJarByClass(STJoin.class);
        job.setMapperClass(Map.class);
        job.setReducerClass(Reduce.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
        FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
        System.exit(job.waitForCompletion(true) ? 0 : 1);

    }

}

export JAVA_HOME=/usr/java/jdk1.7.0_60
export HADOOP_HOME=/cloud/hadoop-2.2.0/etc/hadoop
export PATH=$PATH:$JAVA_HOME/bin:$HADOOP_HOME/bin

尝试平台

一回指令同期修改
hadoop fs -chown supergroup:root /wcout

模仿上题完毕以下内容:对于多个输入文件,即文件math、文件china和文件english,请编写MapReduce程序,对三个文件进行总括平均分,得到三个新的输出文件。下面是输入文件和出口文件的叁个样例供参照他事他说加以考察。
  • 输入文件math.txt的样比方下:
张三    88
李四    99
王五    66
赵六    77
  • 输入文件algs.txt的样举例下:
张三    78
李四    89
王五    96
赵六    67
  • 输入文件english.txt的样举个例子下:
张三    80
李四    82
王五    84
赵六    86
  • 出口文件结果为:
张三    82
李四    90
王五    82
赵六    76

试验代码:

package cn.edu.zucc.mapreduce;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

public class AvgScore {

    public static class Map extends Mapper<LongWritable, Text, Text, IntWritable> {
        @Override
        public void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {
            String line = value.toString();
            String[] nameAndScore = line.split(" ");
            List<String> list = new ArrayList<>(2);
            for (String nameOrScore : nameAndScore) {
                if (!"".equals(nameOrScore)) {
                    list.add(nameOrScore);
                }
            }
            context.write(new Text(list.get(0)), new IntWritable(Integer.parseInt(list.get(1))));
        }
    }

    public static class Reduce extends Reducer<Text, IntWritable, Text, IntWritable> {
        @Override
        public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            int sum = 0;
            int count = 0;
            for (IntWritable value : values) {
                sum  = Integer.parseInt(value.toString());
                count  ;
            }
            int average = sum / count;
            context.write(key, new IntWritable(average));
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "hdfs://localhost:9000");
        String[] otherArgs = new String[]{"input_2", "output_2"};
        if (otherArgs.length != 2) {
            System.err.println("Usage: AvgScore <in> <out>");
            System.exit(2);
        }
        Job job = Job.getInstance(conf, "AvgScore");
        job.setJarByClass(AvgScore.class);
        job.setMapperClass(Map.class);
        job.setReducerClass(Reduce.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);
        FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
        FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}
[root@bogon .ssh]# ls
authorized_keys  id_dsa  id_dsa.pub  known_hosts
[root@bogon .ssh]# 
仿照上题达成以下内容:现存七个输入文件三个文本,三个是工厂名与地址编号的附和关系;另一个是地点编号和地点名的呼应关系。需要从输入数据中搜索工厂名和地址名的相应关系,输出"工厂名——地址名"表。
  • 输入文件factory.txt:
factoryname addressID
Beijing Red Star   1
Shenzhen Thunder   3
Guangzhou Honda   2
Beijing Rising   1
Guangzhou Development Bank    2
Tencent   3
Bank of Beijing   1
  • 输入文件address.txt:
addressID    addressname
1            Beijing
2            Guangzhou
3            Shenzhen
4            Xian
  • 输出文件内容如下:
factoryname addressname
Back of Beijing       Beijing 
Beijing Rising    Beijing 
Beijing Red Star      Beijing 
Guangzhou Development Bank    Guangzhou 
Guangzhou Honda           Guangzhou 
Tencent           Shenzhen 
Shenzhen Thunder          Shenzhen 

尝试代码:

package cn.edu.zucc.mapreduce;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

public class MTJoin {
    public static int time = 0;

    public static class Map extends Mapper<Object, Text, Text, Text> {

        @Override
        protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            if (line.contains("factoryname") || line.contains("addressID")) {
                return;
            }
            String[] strings = line.split(" ");
            List<String> list = new ArrayList<>();
            for (String information : strings) {
                if (!"".equals(information)) {
                    list.add(information);
                }
            }
            String addressID;
            StringBuilder stringBuilder = new StringBuilder();
            if (StringUtils.isNumeric(list.get(0))) {
                addressID = list.get(0);
                for (int i = 1; i < list.size(); i  ) {
                    if (i != 1) {
                        stringBuilder.append(" ");
                    }
                    stringBuilder.append(list.get(i));
                }
                context.write(new Text(addressID), new Text("1 "   stringBuilder.toString()));
            } else {
                addressID = list.get(list.size() - 1);
                for (int i = 0; i < list.size() - 1; i  ) {
                    if (i != 0) {
                        stringBuilder.append(" ");
                    }
                    stringBuilder.append(list.get(i));
                }
                context.write(new Text(addressID), new Text("2 "   stringBuilder.toString()));
            }
        }
    }

    public static class Reduce extends Reducer<Text, Text, Text, Text> {

        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            if (time == 0) {
                context.write(new Text("factoryname"), new Text("addressname"));
                time  ;
            }
            List<String> factory = new ArrayList<>();
            List<String> address = new ArrayList<>();
            for (Text text : values) {
                String s = text.toString();
                String[] relation = s.split("\ ");
                if ("1".equals(relation[0])) {
                    address.add(relation[1]);
                } else {
                    factory.add(relation[1]);
                }
            }
            int factoryNum = factory.size();
            int addressNum = address.size();
            if (factoryNum != 0 && addressNum != 0) {
                for (int m = 0; m < factoryNum; m  ) {
                    for (int n = 0; n < addressNum; n  ) {
                        context.write(new Text(factory.get(m)),
                                new Text(address.get(n)));
                    }
                }
            }
        }

    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "hdfs://localhost:9000");
        String[] ioArgs = new String[]{"input_3", "output_3"};
        String[] otherArgs = new GenericOptionsParser(conf, ioArgs)
                .getRemainingArgs();
        if (otherArgs.length != 2) {
            System.err.println("Usage: Multiple Table Join <in> <out>");
            System.exit(2);
        }
        Job job = Job.getInstance(conf, "Mutiple table join ");
        job.setJarByClass(MTJoin.class);
        job.setMapperClass(Map.class);
        job.setReducerClass(Reduce.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
        FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
        System.exit(job.waitForCompletion(true) ? 0 : 1);

    }

}

2.1 上传hadoop包

  1. 通过试验驾驭基本的MapReduce编制程序方法。
  2. 驾驭用MapReduce消除一部分大范围的数码管理难题,满含数据去重、数据排序和数量发掘等。
  3. 经过操作MapReduce的实验,模仿实验内容,浓厚了然MapReduce的长河,精晓MapReduce程序的编制程序方式。

  默许安装在NameNode 节点上,但那样...不安全

模仿上题实现以下内容:对于五个输入文件,即文件A和文件B,请编写MapReduce程序,对七个公文实行计算单词数量,获得叁个新的出口文件C。下边是输入文件和输出文件的三个样例供参照他事他说加以考察。
  • 输入文件a.txt的样举个例子下:
hello world 
wordcount java
android hbase
hive pig
  • 输入文件b.txt的样比如下:
hello hadoop 
spring mybatis
hive hbase
pig android
  • 出口文件的结果为:
android  2
hadoop    1
hbase      2
hello      2
hive        2
java        1
mybatis  1
pig      2
spring    1
wordcount   1
world      1

实践代码:

package cn.edu.zucc.mapreduce;

import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class WordCount {

    public static class Map extends Mapper<Object, Text, Text, IntWritable> {
        private static final IntWritable one = new IntWritable(1);
        private Text word = new Text();

        @Override
        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            String lineValue = value.toString();
            String[] words = lineValue.split(" ");
            for (String singleWord : words) {
                word.set(singleWord);
                context.write(word, one);
            }

        }
    }

    public static class Reduce extends Reducer<Text, IntWritable, Text, IntWritable> {
        private IntWritable result = new IntWritable();

        @Override
        public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable value : values) {
                sum  = value.get();
            }
            result.set(sum);
            context.write(key, result);
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "hdfs://localhost:9000");
        String[] otherArgs = new String[]{"input_1", "output_1"};
        if (otherArgs.length != 2) {
            System.err.println("Usage: Wordcount <in> <out>");
            System.exit(2);
        }
        Job job = Job.getInstance(conf, "Wordcount");
        job.setJarByClass(WordCount.class);
        job.setMapperClass(Map.class);
        job.setReducerClass(Reduce.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
        FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

<configuration>
<!-- 告诉hadoop以后M福睿斯运转在YAEvoqueN上 YAPRADON是三个能源调节连串,
岂但可以运作MapReduce,仍是能够运行Storm,斯Parker等总结框架 -->
<property>
<name>mapreduce.frameword,name</name>
<value>yarn</value>
</property>
</configuration>

$ ssh-keygen -t dsa -P '' -f ~/.ssh/id_dsa 

tar -zxvf hadoop-2.2.0.tar.gz -C /cloud/

在底行方式能够推行,查看java_home
:echo ${JAVA_HOME}

NameNode

配置hadoop

HDFS的公文模仿Linux上

  实行进度:

  在统一的经过中一旦有新的client操作,则发出新的edits.new文件,不在使用以前的edits文件

hadoop fs -put /root/jdk-7u60-linux-i586.gz hdfs://cloud01:9000/jdk

从 HDFS 下载文件到地点

2.2 解压hadoop包
先是在根目录下创办几个cloud目录

  ③ fstime:保存近日一回checkpoint(还原点)的时间
  (checkpoint是保留近些日子一次做还原点的音讯)

文件的上传

布署201上的免密码登入

元数据存款和储蓄细节

2.8 测试 MR 和 YARN

  HA的一个消除方案.但不帮助热备.配置就能够。

mv mapred-site.xml.template mapred-site.xml

将id_dsa.pub拷贝到就要免密码登入的那台机器上

 合併的触发条件(何时checkpoint):

#格式化只要求贰回
[root@cloud01 bin]# hdfs namenode -format

</configuration>

Live Nodes : 代表DataNode存活的节点

 

2.7 测试HDFS
jps
NameNode: HDFS的可怜,HA(High Ability,高可信性)集群中NameNode是七个

(内部存款和储蓄器一份,磁盘一份)
NameNode(FileName,replicas,block-ids,id2host...)

2.5 开头化文件HDFS(格式化文件系统)

    从NameNode 上下载元数据音讯(fsimage,edits),然后把两个合併,生成新的fsimage,在该地保存,并将其推送到NameNode,替换旧的fsimage

出口文件夹个数 文件个数 大小

2.3 配置hadoop伪布满式(要修改5个布局文件)
/cloud/hadoop-2.2.0/etc/hadoop
修改配置文件

在201上使用  ssh 192.168.1.202:22 登录到202上

配备跨节点的免密码登陆

  Hadoop会维护二个 fsimage 文件,也正是nameNode中metedata的镜像,可是fsimage不会任何时候与namenode
  内部存款和储蓄器中的metedata保存一致,而是每间隔一段时间通过联合edits文件来更新内容(Hadoop 1.0 与 2.0的伪布满式,)
  hadoop2.0会实时进展联合).
  SecondaryNameNode便是用来归并fsimage和edits文件来更新NameNode的metedata的(用新的更迭旧的).

  fs.checkpoint.size 规定edits文件的最大值,一旦超越那个值则强制checkpoint,不管是或不是达到最大时间间距。默许大小是64M。

hadoop fs -chmod -库罗德 -x /wcout (-Enclave 表示递归 x表示试行权限)

  fs.checkpoint.period 钦定两回checkpoint的最大时间隔开分离,暗中同意3600秒

  到有“写乞求” 到来时,nameNode会首先写editlog到磁盘,即向edits文件中写日记,
  成功重返后,才会修改内存,何况向客商端再次回到

亟需先做地点免密码登陆,然后做跨节点免密码登入

SecondaryNameNode:NameNode的副手,完毕数据的联手,但是不是实时的

  • 如假设文件夹表示并不存款和储蓄
    1 如果是文件表示文件的别本个数
    root 所属顾客
    supergroup 所属组
    39343 大小
    终极修改时间
    目录/文件名

问怎样要开展分块存款和储蓄?

配置结果为 201-->202,201-->203, 假诺须求相反,则注重重复下边相反进度

生成id_dsa.pub公钥

mkdir /cloud

-count 总结文件的个数
hadoop fs -count /

各类文件皆有一个校验值,在读取时假使校验值发生变化,表达文件损坏
此处运用CRC32校验算法

ssh-keygen -t dsa -P '' -f ~/.ssh/id_dsa 

  以上这几个文件是保存在linux的文件系统中。

文件 别本数 数据块 每块所在的机械上
test/a.log, 3, {blk_1, blk_2}, [{blk_1: [h0,h1,h3]},{blk_2:[h0,h2,h4]}]

能够经过 jps 的章程注明(jps是java 的指令,在bin下)
还足以经过浏览器的不二秘技阐明
(hdfs管理界面)
(yarn管理分界面)

  ② edits:操作日志文件。
  (用来记录操作日志的音讯)

2.4 将hadoop的bin 增加到遇到变量中,
能够再任何目录下都得以运作

  

</configuration>

在 192.168.1.202 上将id_dsa.pub追加到 authorized_keys 日子文件上

Browse th filesystem: 浏览文件系统
使用主机名 需求在windows上 配置映射关系
c:/windows/system32/dirvers/etc/host

-rw-r--r-- 1 root supergroup 39343 2014-06-18 10:33 /in.log
drwxr-xr-x - root supergroup 0 2014-06-18 10:44 /itcast0106

上传Linux上的公文到HDFS上

vim etc/profile

hdfs dfs -ls /

#格式化后会发生tmp目录

</configuration>

hadoop 2.0

实行下列命令

布局单台机器的免密码登陆

$ cat ~/id_dsa.pub >> ~/.ssh/authorized_keys

DataNode:HDFS的小弟

HDFS块的轻重

ResourceManager:YA兰德安德拉N的那多少个,肩负能源的调节

  是漫天文件系统的军管节点。维护整个文件系统的文书目录树,文件/目录的
  元数据音信和每种文件对应的数额块列表。接收客户的伸手操作。

#有个小标题(供给一再输密码)

hadoop fs -get hdfs://cloud01:9000/jdk /home/jdk1.7

  ① fsimage:元数据镜像文件。存款和储蓄某临时光NameNode内部存储器元数据音讯,保存在磁盘中。
  (内存此中MateDate的镜像文件,内部存款和储蓄器的东西种类化到磁盘中)

上传成功后得以再 (hdfs管理分界面)查看

  NameNode 始终在内部存款和储蓄器中保存metedata,用于拍卖“读央求”

目录下并未有mapred-site.xml,唯有mapred-site.xml.template
急需改名

HDFFS架构

在.ssh目录下转移

免密码登陆可以在肆意一台机器上输入指令,能够运维全体机器上的历程

2.6 启动HDFS和YARN
cd /sbin
./start-all.sh(过时了 This script is Deprecated. Instead use start-dfs.sh
and start-yarn.sh)

先执行

 

  hdfs-site.xml 的 dfs.name.dir 属性

假诺须要递归修改,需求增添 -中华V

hadoop fs -mkdir /cloud 创立目录

.配置免密码登陆

NodeManager:YA凯雷德N的小叔子,负担干活

平价存款和储蓄,MapReduce方便读取

第五个:yarn-site.xml
vim yarn-site.xml

tar: 打包和平解决包
-z:用来管理gz格式
x:代表释放 c:代表创造
v:呈现解压进度详细的情况
f:file
-c:把公文解压到什么样地点

<configuration>
<!-- NodeManager 获取数据的秘诀是 shuffle -->
<property>
<name>yarn.nodemanager.aux-services</name>
<value>mapreduce_shuffle</value>
</property>
<!-- 指定YARN的老大(ResourceManager) 的地址 -->
<property>
<name>yarn.resourcemanager.hostname</name>
<value>cloud01</value>
</property>

hadoop fs -rm 能够去除文件,不可能去除 目录
hadoop fs -rm -r 可以去除目录

hadoop fs -ls -R -h /

在201上生成秘钥 

彩民之家论坛9066777 2

#刷新配置
source etc/profile

$ ssh-keygen -t dsa -P '' -f ~/.ssh/id_dsa 
$ cat ~/.ssh/id_dsa.pub >> ~/.ssh/authorized_keys

<configuration>
<!-- 钦点HDFS保存数据别本的数据 ,如若不配备私下认可别本数为3 -->
<property>
<name>dfs.replication</name>
<value>1</value>
</property>

 

drwxr-xr-x-
(第二位:类型;3位:所属顾客权限;3位:所属组权限;3位:其余客户权限)

<configuration>
<!-- 用来内定HDFS的丰裕(NameNode的地点) -->
<property>
<name>fs.defaultFS</name>
<value>hdfs://cloud01:玖仟</value>(这里写主机名,经过host文件配置过,也足以写ip地址)
</property>
<!-- 用来钦定Hadoop运行时发出文书的寄存目录,配置linux上有些具体的目录 -->
<property>
<name>hadoop.tmp.dir</name>
<value>/cloud/hadoop-2.2.0/tmp</value>
</property>

id_dsa 为私钥,id_dsa.pub为公钥

第二个:core-site.xml
vim core-site.xml

hadoop fs -ls /

第一个:hadoop-env.sh hadoop依赖jdk
vim hadoop-env.sh
export JAVA_HOME=/usr/java/jdk1.7.0_60

hadoop fs -chown supergroup /in.log 修改所属客商
hadoop fs -chgrp root /in.log 修改所属组

#hadoop namenode -format(过时了,但还是能用)

第四个:mapred-site.xml

-奥迪Q5代表递归查看,-h (humanRead)人类可读 将字节变成带单位的 M 或 K 等

递归的铲除实践权限

Hadoop 1.0 块的大小暗许 64M 67108864字节
Hadoop 2.0 块的暗许大小 128M 134217728字节

注:

万一不做免密码登入,须求在每一台机械上输入运行进程命令

文件的下载

SecondaryNameNode

版权声明:本文由彩民之家高手论坛发布于编程技术,转载请注明出处:Hadoop-2.2.0 (传 hadoop-2.2.0.tar.gz)彩民之家论坛9