数据开发系列篇(2):大数据处理和编程实践Hadoop(二)

浏览: 2401

Clipboard Image.png

下面是岑文初师兄总结的Hadoop入门的实践放翁(文初)的一亩三分地,基本上可以作为入门了解去看下,太细节的可以自己去找本书看看。

其实参看Hadoop官方文档已经能够很容易配置分布式框架运行环境了,不过这里既然写了就再多写一点,同时有一些细节需要注意的也说明一下,其实也就是这些细节会让人摸索半天。Hadoop可以单机跑,也可以配置集群跑,单机跑就不需要多说了,只需要按照Demo的运行说明直接执行命令即可。这里主要重点说一下集群配置运行的过程。

环境

7台普通的机器,操作系统都是Linux。内存和CPU就不说了,反正Hadoop一大特点就是机器在多不在精。JDK必须是1.5以上的,这个切记。7台机器的机器名务必不同,后续会谈到机器名对于MapReduce有很大的影响。

部署考虑

正如上面我描述的,对于Hadoop的集群来说,可以分成两大类角色:Master和Slave,前者主要配置NameNode和JobTracker的角色,负责总管分布式数据和分解任务的执行,后者配置DataNode和TaskTracker的角色,负责分布式数据存储以及任务的执行。本来我打算看看一台机器是否可以配置成Master,同时也作为Slave使用,不过发现在NameNode初始化的过程中以及TaskTracker执行过程中机器名配置好像有冲突(NameNode和TaskTracker对于Hosts的配置有些冲突,究竟是把机器名对应IP放在配置前面还是把Localhost对应IP放在前面有点问题,不过可能也是我自己的问题吧,这个大家可以根据实施情况给我反馈)。最后反正决定一台Master,六台Slave,后续复杂的应用开发和测试结果的比对会增加机器配置。

实施步骤

  1. 在所有的机器上都建立相同的目录,也可以就建立相同的用户,以该用户的home路径来做hadoop的安装路径。例如我在所有的机器上都建立了/home/wenchu。
  2. 下载Hadoop,先解压到Master上。这里我是下载的0.17.1的版本。此时Hadoop的安装路径就是/home/wenchu/hadoop-0.17.1。
  3. 解压后进入conf目录,主要需要修改以下文件:hadoop-env.sh,hadoop-site.xml、masters、slaves。

    Hadoop的基础配置文件是hadoop-default.xml,看Hadoop的代码可以知道,默认建立一个Job的时候会建立Job的Config,Config首先读入hadoop-default.xml的配置,然后再读入hadoop-site.xml的配置(这个文件初始的时候配置为空),hadoop-site.xml中主要配置你需要覆盖的hadoop-default.xml的系统级配置,以及你需要在你的MapReduce过程中使用的自定义配置(具体的一些使用例如final等参考文档)。

    以下是一个简单的hadoop-site.xml的配置:

    <?xml version="1.0"?>
    <?xml-stylesheet type="text/xsl" href="https://ask.hellobi.com/configuration.xsl"?>
    <!-- Put site-specific property overrides in this file. -->
    <configuration>
    <property>
    <name>fs.default.name</name>//你的namenode的配置,机器名加端口
    <value>hdfs://10.2.224.46:54310/</value>
    </property>
    <property>
    <name>mapred.job.tracker</name>//你的JobTracker的配置,机器名加端口
    <value>hdfs://10.2.224.46:54311/</value>
    </property>
    <property>
    <name>dfs.replication</name>//数据需要备份的数量,默认是三
    <value>1</value>
    </property>
    <property>
    <name>hadoop.tmp.dir</name>//Hadoop的默认临时路径,这个最好配置,如果在新增节点或者其他情况下莫名其妙的DataNode启动不了,就删除此文件中的tmp目录即可。不过如果删除了NameNode机器的此目录,那么就需要重新执行NameNode格式化的命令。
    <value>/home/wenchu/hadoop/tmp/</value>
    </property>
    <property>
    <name>mapred.child.java.opts</name>//java虚拟机的一些参数可以参照配置
    <value>-Xmx512m</value>
    </property>
    <property>
    <name>dfs.block.size</name>//block的大小,单位字节,后面会提到用处,必须是512的倍数,因为采用crc作文件完整性校验,默认配置512是checksum的最小单元。
    <value>5120000</value>
    <description>The default block size for new files.</description>
    </property>
    </configuration>

    hadoop-env.sh文件只需要修改一个参数:

    # The java implementation to use.  Required.
    export JAVA_HOME=/usr/ali/jdk1.5.0_10

    配置你的Java路径,记住一定要1.5版本以上,免得莫名其妙出现问题。

    Masters中配置Masters的IP或者机器名,如果是机器名那么需要在/etc/hosts中有所设置。Slaves中配置的是Slaves的IP或者机器名,同样如果是机器名需要在/etc/hosts中有所设置。范例如下,我这里配置的都是IP:

    Masters:
    10.2.224.46

    Slaves:
    10.2.226.40
    10.2.226.39
    10.2.226.38
    10.2.226.37
    10.2.226.41
    10.2.224.36

  4. 建立Master到每一台Slave的SSH受信证书。由于Master将会通过SSH启动所有Slave的Hadoop,所以需要建立单向或者双向证书保证命令执行时不需要再输入密码。在Master和所有的Slave机器上执行:ssh-keygen -t rsa。执行此命令的时候,看到提示只需要回车。然后就会在/root/.ssh/下面产生id_rsa.pub的证书文件,通过scp将Master机器上的这个文件拷贝到Slave上(记得修改名称),例如:scp root@masterIP:/root/.ssh/id_rsa.pub /root/.ssh/46_rsa.pub,然后执行cat /root/.ssh/46_rsa.pub >>/root/.ssh/authorized_keys,建立authorized_keys文件即可,可以打开这个文件看看,也就是rsa的公钥作为key,user@IP作为value。此时可以试验一下,从master ssh到slave已经不需要密码了。由slave反向建立也是同样。为什么要反向呢?其实如果一直都是Master启动和关闭的话那么没有必要建立反向,只是如果想在Slave也可以关闭Hadoop就需要建立反向。
  5. 将Master上的Hadoop通过scp拷贝到每一个Slave相同的目录下,根据每一个Slave的Java_HOME的不同修改其hadoop-env.sh。
  6. 修改Master上/etc/profile:
    新增以下内容:(具体的内容根据你的安装路径修改,这步只是为了方便使用)
    export HADOOP_HOME=/home/wenchu/hadoop-0.17.1
    export PATH=$PATH:$HADOOP_HOME/bin

    修改完毕后,执行source /etc/profile来使其生效。
  7. 在Master上执行Hadoop namenode –format,这是第一需要做的初始化,可以看作格式化吧,以后除了在上面我提到过删除了Master上的hadoop.tmp.dir目录,否则是不需要再次执行的。
  8. 然后执行Master上的start-all.sh,这个命令可以直接执行,因为在6中已经添加到了path路径,这个命令是启动hdfs和mapreduce两部分,当然你也可以分开单独启动hdfs和mapreduce,分别是bin目录下的start-dfs.sh和start-mapred.sh。
  9. 检查Master的logs目录,看看Namenode日志以及JobTracker日志是否正常启动。
  10. 检查Slave的logs目录看看Datanode日志以及TaskTracker日志是否正常。
  11. 如果需要关闭,那么就直接执行stop-all.sh即可。

以上步骤就可以启动Hadoop的分布式环境,然后在Master的机器进入Master的安装目录,执行hadoop jar hadoop-0.17.1-examples.jar wordcount输入路径和输出路径,就可以看到字数统计的效果了。此处的输入路径和输出路径都指的是HDFS中的路径,因此你可以首先通过拷贝本地文件系统中的目录到HDFS中的方式来建立HDFS中的输入路径:

hadoop dfs -copyFromLocal /home/wenchu/test-in test-in。其中/home/wenchu/test-in是本地路径,test-in是将会建立在HDFS中的路径,执行完毕以后可以通过hadoop dfs –ls看到test-in目录已经存在,同时可以通过hadoop dfs –ls test-in查看里面的内容。输出路径要求是在HDFS中不存在的,当执行完那个demo以后,就可以通过hadoop dfs –ls 输出路径看到其中的内容,具体文件的内容可以通过hadoop dfs –cat文件名称来查看。

经验总结和注意事项(这部分是我在使用过程中花了一些时间走的弯路):

  1. Master和Slave上的几个conf配置文件不需要全部同步,如果确定都是通过Master去启动和关闭,那么Slave机器上的配置不需要去维护。但如果希望在任意一台机器都可以启动和关闭Hadoop,那么就需要全部保持一致了。
  2. Master和Slave机器上的/etc/hosts中必须把集群中机器都配置上去,就算在各个配置文件中使用的是IP。这个吃过不少苦头,原来以为如果配成IP就不需要去配置Host,结果发现在执行Reduce的时候总是卡住,在拷贝的时候就无法继续下去,不断重试。另外如果集群中如果有两台机器的机器名如果重复也会出现问题。
  3. 如果在新增了节点或者删除节点的时候出现了问题,首先就去删除Slave的hadoop.tmp.dir,然后重新启动试试看,如果还是不行那就干脆把Master的hadoop.tmp.dir删除(意味着dfs上的数据也会丢失),如果删除了Master的hadoop.tmp.dir,那么就需要重新namenode –format。
  4. Map任务个数以及Reduce任务个数配置。前面分布式文件系统设计提到一个文件被放入到分布式文件系统中,会被分割成多个block放置到每一个的DataNode上,默认dfs.block.size应该是64M,也就是说如果你放置到HDFS上的数据小于64,那么将只有一个Block,此时会被放置到某一个DataNode中,这个可以通过使用命令:hadoop dfsadmin –report就可以看到各个节点存储的情况。也可以直接去某一个DataNode查看目录:hadoop.tmp.dir/dfs/data/current就可以看到那些block了。Block的数量将会直接影响到Map的个数。当然可以通过配置来设定Map和Reduce的任务个数。Map的个数通常默认和HDFS需要处理的blocks相同。也可以通过配置Map的数量或者配置minimum split size来设定,实际的个数为:max(min(block_size,data/#maps),min_split_size)。Reduce可以通过这个公式计算:0.95*num_nodes*mapred.tasktracker.tasks.maximum。

总的来说出了问题或者启动的时候最好去看看日志,这样心里有底。

Hadoop中的命令(Command)总结

这部分内容其实可以通过命令的Help以及介绍了解,我主要侧重于介绍一下我用的比较多的几个命令。Hadoop dfs 这个命令后面加参数就是对于HDFS的操作,和Linux操作系统的命令很类似,例如:

  • Hadoop dfs –ls就是查看/usr/root目录下的内容,默认如果不填路径这就是当前用户路径;
  • Hadoop dfs –rmr xxx就是删除目录,还有很多命令看看就很容易上手;
  • Hadoop dfsadmin –report这个命令可以全局的查看DataNode的情况;
  • Hadoop job后面增加参数是对于当前运行的Job的操作,例如list,kill等;
  • Hadoop balancer就是前面提到的均衡磁盘负载的命令。

其他就不详细介绍了。

————————————转载请注明原文链接————————————————

##知乎专栏:数据分析侠数据分析侠 - 知乎专栏

##微信公众号:数据分析联盟(datafa)

##时间:2016-4-27

Hadoop基本流程

Clipboard Image.png

Clipboard Image.png

一个图片太大了,只好分割成为两部分。根据流程图来说一下具体一个任务执行的情况。

  1. 在分布式环境中客户端创建任务并提交。
  2. InputFormat做Map前的预处理,主要负责以下工作:
    1. 验证输入的格式是否符合JobConfig的输入定义,这个在实现Map和构建Conf的时候就会知道,不定义可以是Writable的任意子类。
    2. 将input的文件切分为逻辑上的输入InputSplit,其实这就是在上面提到的在分布式文件系统中blocksize是有大小限制的,因此大文件会被划分为多个block。
    3. 通过RecordReader来再次处理inputsplit为一组records,输出给Map。(inputsplit只是逻辑切分的第一步,但是如何根据文件中的信息来切分还需要RecordReader来实现,例如最简单的默认方式就是回车换行的切分)
  3. RecordReader处理后的结果作为Map的输入,Map执行定义的Map逻辑,输出处理后的key和value对应到临时中间文件。
  4. Combiner可选择配置,主要作用是在每一个Map执行完分析以后,在本地优先作Reduce的工作,减少在Reduce过程中的数据传输量。
  5. Partitioner可选择配置,主要作用是在多个Reduce的情况下,指定Map的结果由某一个Reduce处理,每一个Reduce都会有单独的输出文件。(后面的代码实例中有介绍使用场景)
  6. Reduce执行具体的业务逻辑,并且将处理结果输出给OutputFormat。
  7. OutputFormat的职责是,验证输出目录是否已经存在,同时验证输出结果类型是否如Config中配置,最后输出Reduce汇总后的结果。

业务场景和代码范例

业务场景描述:可设定输入和输出路径(操作系统的路径非HDFS路径),根据访问日志分析某一个应用访问某一个API的总次数和总流量,统计后分别输出到两个文件中。这里仅仅为了测试,没有去细分很多类,将所有的类都归并于一个类便于说明问题。

Clipboard Image.png

测试代码类图

LogAnalysiser就是主类,主要负责创建、提交任务,并且输出部分信息。内部的几个子类用途可以参看流程中提到的角色职责。具体地看看几个类和方法的代码片断:

LogAnalysiser::MapClass

    public static class MapClass extends MapReduceBase
implements Mapper<LongWritable, Text, Text, LongWritable>
{
public void map(LongWritable key, Text value, OutputCollector<Text, LongWritable> output, Reporter reporter)
throws IOException
{
String line = value.toString();//没有配置RecordReader,所以默认采用line的实现,key就是行号,value就是行内容
if (line == null || line.equals(""))
return;
String[] words = line.split(",");
if (words == null || words.length < 8)
return;
String appid = words[1];
String apiName = words[2];
LongWritable recbytes = new LongWritable(Long.parseLong(words[7]));
Text record = new Text();
record.set(new StringBuffer("flow::").append(appid)
.append("::").append(apiName).toString());
reporter.progress();
output.collect(record, recbytes);//输出流量的统计结果,通过flow::作为前缀来标示。
record.clear();
record.set(new StringBuffer("count::").append(appid).append("::").append(apiName).toString());
output.collect(record, new LongWritable(1));//输出次数的统计结果,通过count::作为前缀来标示
}
}

LogAnalysiser:: PartitionerClass

    public static class PartitionerClass implements Partitioner<Text, LongWritable>
{
public int getPartition(Text key, LongWritable value, int numPartitions)
{
if (numPartitions >= 2)//Reduce 个数,判断流量还是次数的统计分配到不同的Reduce
if (key.toString().startsWith("flow::"))
return 0;
else
return 1;
else
return 0;
}
public void configure(JobConf job){}
}

LogAnalysiser:: CombinerClass

参看ReduceClass,通常两者可以使用一个,不过这里有些不同的处理就分成了两个。在ReduceClass中蓝色的行表示在CombinerClass中不存在。

LogAnalysiser:: ReduceClass

    public static class ReduceClass extends MapReduceBase
implements Reducer<Text, LongWritable,Text, LongWritable>
{
public void reduce(Text key, Iterator<LongWritable> values,
OutputCollector<Text, LongWritable> output, Reporter reporter)throws IOException
{
Text newkey = new Text();
newkey.set(key.toString().substring(key.toString().indexOf("::")+2));
LongWritable result = new LongWritable();
long tmp = 0;
int counter = 0;
while(values.hasNext())//累加同一个key的统计结果
{
tmp = tmp + values.next().get();

counter = counter +1;//担心处理太久,JobTracker长时间没有收到报告会认为TaskTracker已经失效,因此定时报告一下
if (counter == 1000)
{
counter = 0;
reporter.progress();
}
}
result.set(tmp);
output.collect(newkey, result);//输出最后的汇总结果
}
}

LogAnalysiser

	public static void main(String[] args)
{
try
{
run(args);
} catch (Exception e)
{
e.printStackTrace();
}
}
public static void run(String[] args) throws Exception
{
if (args == null || args.length <2)
{
System.out.println("need inputpath and outputpath");
return;
}
String inputpath = args[0];
String outputpath = args[1];
String shortin = args[0];
String shortout = args[1];
if (shortin.indexOf(File.separator) >= 0)
shortin = shortin.substring(shortin.lastIndexOf(File.separator));
if (shortout.indexOf(File.separator) >= 0)
shortout = shortout.substring(shortout.lastIndexOf(File.separator));
SimpleDateFormat formater = new SimpleDateFormat("yyyy.MM.dd");
shortout = new StringBuffer(shortout).append("-")
.append(formater.format(new Date())).toString();


if (!shortin.startsWith("/"))
shortin = "/" + shortin;
if (!shortout.startsWith("/"))
shortout = "/" + shortout;
shortin = "/user/root" + shortin;
shortout = "/user/root" + shortout;
File inputdir = new File(inputpath);
File outputdir = new File(outputpath);
if (!inputdir.exists() || !inputdir.isDirectory())
{
System.out.println("inputpath not exist or isn't dir!");
return;
}
if (!outputdir.exists())
{
new File(outputpath).mkdirs();
}

JobConf conf = new JobConf(new Configuration(),LogAnalysiser.class);//构建Config
FileSystem fileSys = FileSystem.get(conf);
fileSys.copyFromLocalFile(new Path(inputpath), new Path(shortin));//将本地文件系统的文件拷贝到HDFS中

conf.setJobName("analysisjob");
conf.setOutputKeyClass(Text.class);//输出的key类型,在OutputFormat会检查
conf.setOutputValueClass(LongWritable.class); //输出的value类型,在OutputFormat会检查
conf.setMapperClass(MapClass.class);
conf.setCombinerClass(CombinerClass.class);
conf.setReducerClass(ReduceClass.class);
conf.setPartitionerClass(PartitionerClass.class);
conf.set("mapred.reduce.tasks", "2");//强制需要有两个Reduce来分别处理流量和次数的统计
FileInputFormat.setInputPaths(conf, shortin);//hdfs中的输入路径
FileOutputFormat.setOutputPath(conf, new Path(shortout));//hdfs中输出路径

Date startTime = new Date();
System.out.println("Job started: " + startTime);
JobClient.runJob(conf);
Date end_time = new Date();
System.out.println("Job ended: " + end_time);
System.out.println("The job took " + (end_time.getTime() - startTime.getTime()) /1000 + " seconds.");
//删除输入和输出的临时文件
fileSys.copyToLocalFile(new Path(shortout),new Path(outputpath));
fileSys.delete(new Path(shortin),true);
fileSys.delete(new Path(shortout),true);
}

以上的代码就完成了所有的逻辑性代码,然后还需要一个注册驱动类来注册业务Class为一个可标示的命令,让hadoop jar可以执行。

public class ExampleDriver {
public static void main(String argv[]){
ProgramDriver pgd = new ProgramDriver();
try {
pgd.addClass("analysislog", LogAnalysiser.class, "A map/reduce program that analysis log .");
pgd.driver(argv);
}
catch(Throwable e){
e.printStackTrace();
}
}
}

将代码打成jar,并且设置jar的mainClass为ExampleDriver这个类。在分布式环境启动以后执行如下语句:

hadoop jar analysiser.jar analysislog /home/wenchu/test-in /home/wenchu/test-out

在/home/wenchu/test-in中是需要分析的日志文件,执行后就会看见整个执行过程,包括了Map和Reduce的进度。执行完毕会在/home/wenchu/test-out下看到输出的内容。有两个文件:part-00000和part-00001分别记录了统计后的结果。 如果需要看执行的具体情况,可以看在输出目录下的_logs/history/xxxx_analysisjob,里面罗列了所有的Map,Reduce的创建情况以及执行情况。在运行期也可以通过浏览器来查看Map,Reduce的情况:http://MasterIP:50030/jobtracker.jsp

Hadoop集群测试

首先这里使用上面的范例作为测试,也没有做太多的优化配置,这个测试结果只是为了看看集群的效果,以及一些参数配置的影响。

文件复制数为1,blocksize 5M

Slave数处理记录数(万条)执行时间(秒)295382950337495244950178695216950114

Blocksize 5M

Slave数处理记录数(万条)执行时间(秒)2(文件复制数为1)9503372(文件复制数为3)9503396(文件复制数为1)9501146(文件复制数为3)950117

文件复制数为1

Slave数处理记录数(万条)执行时间(秒)6(blocksize 5M)95216(blocksize 77M)95264(blocksize 5M)9501784(blocksize 50M)950546(blocksize 5M)9501146(blocksize 50M)950446(blocksize 77M)95074

测试的数据结果很稳定,基本测几次同样条件下都是一样。通过测试结果可以看出以下几点:

  1. 机器数对于性能还是有帮助的(等于没说^_^)。
  2. 文件复制数的增加只对安全性有帮助,但是对于性能没有太多帮助。而且现在采取的是将操作系统文件拷贝到HDFS中,所以备份多了,准备的时间很长。
  3. blocksize对于性能影响很大,首先如果将block划分的太小,那么将会增加job的数量,同时也增加了协作的代价,降低了性能,但是配置的太大也会让job不能最大化并行处理。所以这个值的配置需要根据数据处理的量来考虑。
  4. 最后就是除了这个表里面列出来的结果,应该去仔细看输出目录中的_logs/history中的xxx_analysisjob这个文件,里面记录了全部的执行过程以及读写情况。这个可以更加清楚地了解哪里可能会更加耗时。

随想

“云计算”热的烫手,就和SAAS、Web2及SNS等一样,往往都是在搞概念,只有真正踏踏实实的大型互联网公司,才会投入人力物力去研究符合自己的分布式计算。其实当你的数据量没有那么大的时候,这种分布式计算也就仅仅只是一个玩具而已,只有在真正解决问题的过程中,它深层次的问题才会被挖掘出来。

这几篇文章(分布式计算开源框架Hadoop介绍,Hadoop中的集群配置和使用技巧)仅仅是为了给对分布式计算有兴趣的朋友抛个砖,要想真的掘到金子,那么就踏踏实实的去用、去想、去分析。或者自己也会更进一步地去研究框架中的实现机制,在解决自己问题的同时,也能够贡献一些什么。

前几日看到有人跪求成为架构师的方式,看了有些可悲,有些可笑,其实有多少架构师知道什么叫做架构?架构师的职责是什么?与其追求这么一个名号,还不如踏踏实实地做块石头沉到水底。要知道,积累和沉淀的过程就是一种成长。

推荐 0
本文由 面包君 创作,采用 知识共享署名-相同方式共享 3.0 中国大陆许可协议 进行许可。
转载、引用前需联系作者,并署名作者且注明文章出处。
本站文章版权归原作者及原出处所有 。内容为作者个人观点, 并不代表本站赞同其观点和对其真实性负责。本站是一个个人学习交流的平台,并不用于任何商业目的,如果有任何问题,请及时联系我们,我们将根据著作权人的要求,立即更正或者删除有关内容。本站拥有对此声明的最终解释权。

0 个评论

要回复文章请先登录注册