hadoop初遇爬坑

下载

在下面的地址下载字节须要的版本，若是是windows下建议先下载hadoop windows工具包看一下如今支持哪些版本，而后选择对应的hadoop版本。

hadoop下载 hadoop windows工具包

下载完以后解压：

hadoop windows工具包可选版本

hadoop windows工具包bin目录

hadoop根目录

hadoop bin目录

使用hadoop工具包的bin目录覆盖hadoop的bin目录，注意对应版本，通常大版本对上基本就没有问题。覆盖以前最好先备份。

配置文件

core-site.xml

<configuration>
    <property>       
        <name>fs.defaultFS</name>       
        <value>hdfs://localhost:9000</value>   
    </property>
    <property>
        <name>hadoop.tmp.dir</name>
        <value>/G:/datacenter/data30/tmp</value>
    </property>
</configuration>

主要配置的是hdfs这个文件系统的访问接口和临时目录。

hdfs-site.xml

<configuration>
    <property>       
        <name>dfs.replication</name>       
        <value>1</value>   
    </property>   
    <property>       
        <name>dfs.namenode.name.dir</name>       
        <value>/G:/datacenter/data30/namenode</value>   
    </property>   
    <property>       
        <name>dfs.datanode.data.dir</name>     
        <value>/G:/datacenter/data30/datanode</value>   
    </property>
</configuration>

配置hdfs系统的namenode目录和datanode目录。

mapred-site.xml

<configuration>   
    <property>       
        <name>mapreduce.framework.name</name>       
        <value>yarn</value>   
    </property>
</configuration>

yarn-site.xml

<configuration>   
    <property>       
        <name>yarn.nodemanager.aux-services</name>       
        <value>mapreduce_shuffle</value>   
    </property>   
    <property>       
        <name>yarn.nodemanager.aux-services.mapreduce.shuffle.class</name>   
        <value>org.apache.hadoop.mapred.ShuffleHandler</value>   
    </property>
</configuration>

这里的yarn不是打包工具，而是一个资源调度器，YARN只提供运算资源的调度，用户程序向YARN申请资源，YARN就负责分配资源。

YARN中的主管角色叫ResourceManager，YARN中具体提供运算资源的角色叫NodeManager。

启动

把winutil的bin目录拷贝到hadoop的bin目录下，直接执行替换操做，建议替换以前先备份。而后把hadoop.dll动态连接文件拷贝一份到C:\Windows\System32目录下。

若是配置的JAVA_HOME中在C:\Program Files目录下，要修改一下hadoop-env.cmd：

set JAVA_HOME=%JAVA_HOME%

set JAVA_HOME=C:\PROGRA~1\Java\jdk1.8.0_191

用下面一个的替换上面一个，PROGRA~1代替的是Program Files，由于Program Files包含空格。

替换完成以后先配置一下环境变量，假设已经配置了JAVA的环境变量，就只须要配置HADOOP_HOME，指向hadoop的解压目录就能够了，而后path中加上hadoop的bin和sbin目录。

记得前面建立的目录，如今要先格式化hadoop：

hadoop namenode -format
hdfs namenode -format

上面2个命令任意一个，推荐下面一个，格式化成功以后，前面配置的namenode目录下会出现一个current文件夹。

如今就能够执行脚本：

start-all.cmd

固然也能够前后执行:

start-hdfs.xml
start-yarn.xml

start-hdfs.xml启动的是namenode和datanode，start-yarn.xml启动的是resourcemanager和nodemanager。

启动好以后执行一下jps命令，可能看到下面的内容：

JobTracker
SecondaryNameNode
NodeManager
ResourceManager
NameNode
DataNode

由于是单机，不是集群因此可能只有下面4个：

NodeManager
ResourceManager
NameNode
DataNode

能够访问： http://localhost:50070 来查看hdfs的web界面。

hadoop3.0开始访问： http://localhost:9870

也能够经过下面的配置修改(core-site.xml)：

<property>
  <name>dfs.namenode.http-address</name>
  <value>127.0.0.1:50070</value>
</property>

能够访问： http://localhost:8088 来查看yarn的web界面，为了先用起来这里先不详细介绍，后面会补充一点内容。

hdfs命令与Java接口

hdfs基本命令

hadoop fs -rm -r /dir #删除
hadoop fs -ls -R /dir #列表
hadoop fs -mkdir /dir #建立目录
hadoop fs -put in.txt /tmp #上传
hadoop fs -get /tmp/in.txt out.txt #下载

咱们能够看到hadoop fs命令和操做linux的命令基本一致，上面须要注意一点的就是文件上传和下载了。上传文件指定的目录必定要存在。

例如，上面的上传命令就会把当前目录下的in.txt文件上传到hdfs的/tmp目录下，使用列表命令就能够看到hdfs有一个/tmp/in.txt文件。

下载命令也同样，指定hdfs的路径，后一个是本地的路径。

hdfs Java接口

这里使用maven地方方式，先添加依赖：

<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-common</artifactId>
    <version>${hadoop.version}</version>
</dependency>
<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-hdfs</artifactId>
    <version>${hadoop.version}</version>
</dependency>
<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-client</artifactId>
    <version>${hadoop.version}</version>
</dependency>
<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-mapreduce-client-core</artifactId>
    <version>${hadoop.version}</version>
</dependency>

这里的${hadoop.version}选择对应hadoop的版本就能够了。

下面是一个hdfs操做的简单示例：

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import org.junit.Test;

import java.io.FileInputStream;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

public class HDFSUtilTest {

    private static final String PATH = "hdfs://localhost:9000/";

    private static final String DIR = "/dir";

    private static final String FILE = "/dir/hello2";

    public static final String IN = "G:\\tmp\\in2.txt";


    /**
     * 删除文件夹 hadoop fs -rmr /dir
     * @throws IOException
     */
    @Test
    public void remove() throws IOException, URISyntaxException {
        FileSystem fileSystem = getFileSystem();
        Path path = new Path(DIR);
        fileSystem.delete(path, true);
    }

    /**
     * 浏览文件夹 hadoop fs -lsr path
     * @throws IOException
     * @throws URISyntaxException
     */
    @Test
    public void list() throws IOException, URISyntaxException {
        FileSystem fileSystem = getFileSystem();
        Path root = new Path("/");
        FileStatus[] listStatus = fileSystem.listStatus(root);
        for (FileStatus fileStatus : listStatus) {
            String isDir = fileStatus.isDirectory() ? "文件夹" : "文件";
            String permission = fileStatus.getPermission().toString();
            int replication = fileStatus.getReplication();
            long len = fileStatus.getLen();
            String path = fileStatus.getPath().toString();
            System.out.println(isDir + "\t" + permission + "\t" + replication
                    + "\t" + len + "\t" + path);
        }
    }

    /**
     * 下载文件 hadoop fs -get src des
     * @throws IOException
     * @throws URISyntaxException
     */
    @Test
    public void getData() throws IOException, URISyntaxException {
        FileSystem fileSystem = getFileSystem();
//        String file = "/out/_SUCCESS";
        String file = "/out/part-r-00000";
        Path path = new Path(file);
        FSDataInputStream inputStream = fileSystem.open(path);
        IOUtils.copyBytes(inputStream, System.out, 1024, true);
    }

    /**
     * 上传文件 hadoop fs -put src des
     * @throws IOException
     * @throws URISyntaxException
     */
    @Test
    public void putData() throws IOException, URISyntaxException {
        FileSystem fileSystem = getFileSystem();
        Path path = new Path(FILE);
        FSDataOutputStream out = fileSystem.create(path);
        FileInputStream in = new FileInputStream(IN);
        IOUtils.copyBytes(in, out, 1024, true);
    }

    /**
     * 建立文件夹 hadoop fs -mkdir /dir
     * @throws IOException
     * @throws URISyntaxException
     */
    @Test
    public void mkDir() throws IOException, URISyntaxException {
        FileSystem fileSystem = getFileSystem();
        Path path = new Path(DIR);
        fileSystem.mkdirs(path);
    }

    private static FileSystem getFileSystem() throws IOException, URISyntaxException {
        URI uri = new URI(PATH);
        Configuration conf = new Configuration();
        FileSystem fileSystem = FileSystem.get(uri, conf);
        return fileSystem;
    }

}

hadoop mapreduce

下面是一个经典的hadoop mapreduce入门级示例。

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;
import org.apache.hadoop.util.GenericOptionsParser;

import java.io.IOException;
import java.util.StringTokenizer;

public class WordCount {

    public static class TokenizerMapper extends Mapper<Object, Text,Text, IntWritable>{

        private final static IntWritable one = new IntWritable(1);

        private Text word = new Text();

        @Override
        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            StringTokenizer stringTokenizer = new StringTokenizer(value.toString());
            while (stringTokenizer.hasMoreTokens()){
                word.set(stringTokenizer.nextToken());
                context.write(word,one);
            }
        }
    }

    public static class IntSumReducer 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 num : values){
                sum += num.get();
            }
            result.set(sum);
            context.write(key,result);
        }
    }

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration configuration = new Configuration();
        configuration.set("fs.default.name", "hdfs://localhost:9000");
        String[] remainingArgs = new GenericOptionsParser(configuration, args).getRemainingArgs();
        if(remainingArgs.length < 2){
            System.out.println("args error");
            System.exit(2);
        }
        Job job = Job.getInstance(configuration, "word count");
        job.setJarByClass(WordCount.class);
        job.setMapperClass(TokenizerMapper.class);
        job.setCombinerClass(IntSumReducer.class);
        job.setReducerClass(IntSumReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        FileInputFormat.addInputPath(job, new Path(remainingArgs[0]));
        FileOutputFormat.setOutputPath(job, new Path(remainingArgs[1]));
        System.exit(job.waitForCompletion(true) ? 0:1);
    }
}

导包的时候会发现有mapreduce包和mapred包，暂时尚未了解这2个包有什么区别。

main接收2个参数，第一个参数是要处理文件的目录，第二个是处理结果的输出目录。

就是把第一个参数指定的目录下的文件使用TokenizerMapper这个类的map方法，就是使用StringTokenizer分词，而后使用IntSumReducer的reduce方法执行了一个汇总操做。

注意：这个mapreduce操做都在hdfs上进行的，因此能够先经过命令将要分析的文件上传到hdfs上。

另外指定目录的时候是否是以/开头的很重要，若是不是以/开头就会在前面拼接上系统属性user.dir的值。例如，指定的路径是"in/data.txt"，实际上会转换为/home/username/in/data.txt这个路径，若是是"/in/data.txt"，那么就是路径自己。

YARN

Scheduler

Scheduler是调度器，根据应用程序的资源需求进行资源分配，不参与应用程序具体的执行和监控等工做资源分配的单位就是Container，调度器是一个可插拔的组件，用户能够根据本身的需求实 现本身的调度器。YARN 自己为咱们提供了多种直接可用的调度器，好比 FIFO，Fair Scheduler 和Capacity Scheduler等。

ResourceManager

ResourceManager 是基于应用程序对集群资源的需求进行调度的 YARN 集群主控节点，负责 协调和管理整个集群（全部 NodeManager）的资源，响应用户提交的不一样类型应用程序的 解析，调度，监控等工做。ResourceManager 会为每个 Application 启动一个 MRAppMaster，而且MRAppMaster分散在各个NodeManager节点。

ResourceManager的职责：

1. 处理客户端请求
2. 启动或监控 MRAppMaster
3. 监控 NodeManager
4. 资源的分配与调度

NodeManager

NodeManager是YARN集群当中真正资源的提供者，是真正执行应用程序的容器的提供者， 监控应用程序的资源使用状况，并经过心跳向集群资源调度器 ResourceManager 进行汇报以更新本身的健康状态。同时其也会监督Container的生命周期管理，监控每一个 Container 的资源使用状况，追踪节点健康情况，管理日 志和不一样应用程序用到的附属服务。

NodeManager的职责：

1. 管理单个节点上的资源
2. 处理来自 ResourceManager 的命令
3. 处理来自 MRAppMaster 的命令

MRAppMaster

MRAppMaster 对应一个应用程序，职责是：向资源调度器申请执行任务的资源容器，运行 任务，监控整个任务的执行，跟踪整个任务的状态，处理任务失败以异常状况

Container

Container 容器是一个抽象出来的逻辑资源单位。容器是由 ResourceManager Scheduler 服务 动态分配的资源构成，它包括了该节点上的必定量 CPU，内存，磁盘，网络等信息，MapReduce 程序的全部 Task 都是在一个容器里执行完成的，容器的大小是能够动态调整的

ASM

应用程序管理器 ASM 负责管理整个系统中全部应用程序，包括应用程序提交、与调度器协 商资源以启动 MRAppMaster、监控MRAppMaster运行状态并在失败时从新启动它等。

文档

官方文档