Hadoop大数据平台架构

hadoop

hadoop的安装

# 解压hadoop-1.1.2.tar.gz到/opt目录
tar zxvf hadoop-1.1.2.tar.gz
# 修改4个配置文件
cd /opt/hadoop-1.1.2/conf

hadoop-env.sh
修改JAVA_HOME,我们可以使用echo $JAVA_HOME命令来找到。
core-site.xml

<?xml version="1.0"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<!-- Put site-specific property overrides in this file. -->
<configuration>
<property>
  <name>dfs.name.dir</name>
  <value>/hadoop/name</value>
</property>
<property>
  <name>hadoop.tmp.dir</name>
  <value>/hadoop</value>
</property>
<property>
  <name>fs.default.name</name>
   <value>hdfs://test:9000</value>
</property>
</configuration>

hdfs-site.xml

<?xml version="1.0"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<!-- Put site-specific property overrides in this file. -->
<configuration>
<property>
  <name>dfs.data.dir</name>
  <value>/hadoop/data</value>
</property>
</configuration>

mapred-site.xml

<?xml version="1.0"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<!-- Put site-specific property overrides in this file. -->
<configuration>
<property>
  <name>mapred.job.tracker</name>
  <value>test:9001</value>
</property>
</configuration>

编辑/etc/profile设置HADOOP_HOME,并在PATH中追加hadoop下的bin目录。使用source命令生效。

在任意目录输入hadoop可以测试hadoop是否安装成功。

对namenode进行格式化：

hadoop namenode -format

执行start-all.sh，会依次启动namenode、datanode、secondarynamenode、jobtracker、tasktracker，期间需要输入root密码。我们可以使用jps查看hadoop是否正常运行。

HDFS简介

Block：在操作系统中常常是固定大小的逻辑单元。在HDFS中所有的文件分块存储，HDFS中默认的块的大小时64MB，是文件存储和处理的逻辑单元。

HDFS中有2类节点：NameNode和DataNode。其中NameNode是管理节点，存放文件元数据：
1. 文件与数据块的映射表
2. 数据块和数据节点之间的映射表

DataNode是HDFS的管理节点，存放数据块

HDFS数据管理和容错

数据块有多个冗余，DataNode会定期向NameNode发送心跳消息，称之为心跳检测。二级NameNode定期同步元数据映像文件和修复日志，当NameNode故障时备胎转正。

HDFS文件的读写操作

HDFS特点

• 数据冗余，硬件容错
• 流式数据访问，数据一次写入多次读取(写入之后无法更改，除非将块删除后重写)
• 适合存储大文件，如果存储大量的小文件，NameNode的压力会非常大

HDFS适用性和局限性

• 适合数据批量读写，吞吐量高
• 不适合做交互式应用，低延迟很难满足
• 适合一次写入多次读取，顺序读写
• 不支持多用户并发写相同的文件

HDFS的使用

# 列出HDFS上的文件
hadoop fs -ls /
# 将os上的文件放到HDFS上
hadoop fs -put redis-3.0.5.tar.gz /hadoop/user/root/input
# 查看刚刚放置的文件
hadoop fs -ls /hadoop/user/root/input
# 创建目录
hadoop fs -mkdir tmp
# 查看hdfs上的文件
hadoop fs -put update.js tmp
hadoop fs -cat /user/root/tmp/update.js
# 从hdfs上下载文件
hadoop fs -get tmp/update.js local.js
# 显示hdfs状态
hadoop dfsadmin -report

MapReduce原理

MapReduce简而言之就是一种分治思想，将一个大的任务分成多个小的子任务（Map）执行，最后合并子任务的结果（Reduce）。

例如从100GB的网站日志中找出访问次数最多的IP。我们可以将日志文件进行分割，然后分别统计每个分片ip和访问次数，将每个分片的ip进行规约，根据IP来进行hash映射.

MapReduce的运行流程

一个Job会被拆分成多个Task，Task又被分为MapTask和ReduceTask。

• JobTracker的角色是作业调度，分配任务，监控任务执行进度、监控TaskTracker的状态
• TaskTracker的任务是执行任务并向JobTracker汇报任务执行状态。

MapReduce的执行过程

MapReduce容错机制

• 任务失败后重试4次
• 推测执行，因为所有的Reduce任务必须等待Map任务执行完成，所以如果某个MapTaskTracker算的特别慢就会拖后腿，此时会启动另外一个MapTaskTracker与算的慢的那个执行同样的任务，其中任意一个TaskTracker执行完成就会终止另外一个TaskTracker。

wordcount计数程序

计算出文件中每个单词出现的频数，输入结果按照字母顺序进行排序.

编写wordcount.java:

import java.io.IOException;
import java.util.StringTokenizer;
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 WordCount {
    public static class WordCountMap extends
            Mapper<LongWritable, Text, Text, IntWritable> {
        private final IntWritable one = new IntWritable(1);
        private Text word = new Text();
        public void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {
            String line = value.toString();
            StringTokenizer token = new StringTokenizer(line);
            while (token.hasMoreTokens()) {
                word.set(token.nextToken());
                context.write(word, one);
            }
        }
    }
    public static class WordCountReduce extends
            Reducer<Text, IntWritable, Text, IntWritable> {
        public void reduce(Text key, Iterable<IntWritable> values,
                Context context) throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable val : values) {
                sum += val.get();
            }
            context.write(key, new IntWritable(sum));
        }
    }
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = new Job(conf);
        job.setJarByClass(WordCount.class);
        job.setJobName("wordcount");
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        job.setMapperClass(WordCountMap.class);
        job.setReducerClass(WordCountReduce.class);
        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        job.waitForCompletion(true);
    }
}

在linux中进行编译：

mkdir word_count_class
javac -classpath /opt/hadoop-1.1.2/hadoop-core-1.1.2.jar:/opt/hadoop-1.1.2/lib/commons-cli-1.2.jar -d word_count_class WordCount.java
# 进入编译输出的class目录，发现以下的3个文件
WordCount.class  WordCount$WordCountMap.class  WordCount$WordCountReduce.class
# 将以上的3个class文件打包
jar -cvf wordcount.jar *.class

在本地新建一个文件~/input/file1，内容是：

hello world
hello hadoop
hadoop file system
hadoop java api
hello java

再新建一个文件~/input/file2，内容是：

new file
hadoop file
hadoop new world
hadoop free home
hadoop free school

将以上的2个文件提交给hdfs

hadoop fs -mkdir input_wordcount && hadoop fs -put ~/input/* input_wordcount
# 查看是否已经成功上传
hadoop fs -ls input_wordcount

提交并查看任务

hadoop jar word_count_class/wordcount.jar WordCount input_wordcount output_wordcount
# 查看结果
hadoop fs -ls output_wordcount
# 成功将两个文件中的单词个数统计了出来并按照字典序进行了排列
hadoop fs -cat output_wordcount/part-r-00000

使用MapReduce进行数据排序