Hadoop简明教程

本文帮助大家快速上手Hadoop。

关于Hadoop

Google通过三篇重量级论文为大数据时代提供了三项革命性技术：GFS、MapReduce和BigTable，即所谓的Google大数据的“三驾马车”。

• GFS（Google File System）是Google面向大规模数据密集型应用的、可伸缩的分布式文件系统，可在廉价的硬件上运行，并具有可靠的容错能力。
• MapReduce是一种并行编程模式，可以在超大分布式集群上并行运算，对超大规模数据集进行处理。
• BigTable是在GFS上构建的处理结构化数据的分布式数据库，可以用于处理海量数据的更新和随机查询。
  Hadoop和Hbase是基于这三项技术发展出的开源实现。在大数据分析和处理领域，Hadoop兼容体系已经成为一个非常成熟的生态圈，涵盖了很多大数据相关的基础组件，包括Hadoop、Hbase、Hive、Spark、Flink、Storm、Presto、Impala等。

Hadoop拓扑结构

Namenode 和 Datanode

HDFS采用master/slave架构。一个HDFS集群是由一个Namenode和一定数目的Datanodes组成。Namenode是一个中心服务器，负责管理文件系统的名字空间(namespace)以及客户端对文件的访问。集群中的Datanode一般是一个节点一个，负责管理它所在节点上的存储。HDFS暴露了文件系统的名字空间，用户能够以文件的形式在上面存储数据。从内部看，一个文件其实被分成一个或多个数据块，这些块存储在一组Datanode上。Namenode执行文件系统的名字空间操作，比如打开、关闭、重命名文件或目录。它也负责确定数据块到具体Datanode节点的映射。Datanode负责处理文件系统客户端的读写请求。在Namenode的统一调度下进行数据块的创建、删除和复制。

Namenode和Datanode被设计成可以在普通的商用机器上运行。这些机器一般运行着GNU/Linux操作系统(OS)。HDFS采用Java语言开发，因此任何支持Java的机器都可以部署Namenode或Datanode。由于采用了可移植性极强的Java语言，使得HDFS可以部署到多种类型的机器上。一个典型的部署场景是一台机器上只运行一个Namenode实例，而集群中的其它机器分别运行一个Datanode实例。这种架构并不排斥在一台机器上运行多个Datanode，只不过这样的情况比较少见。

集群中单一Namenode的结构大大简化了系统的架构。Namenode是所有HDFS元数据的仲裁者和管理者，这样，用户数据永远不会流过Namenode。

基本管理

Hadoop集群的启动和停止涉及到其多个服务的管理，主要包括HDFS（Hadoop Distributed File System）和YARN（Yet Another Resource Negotiator）。这些服务通常通过位于Hadoop安装目录下的脚本来控制。

启动Hadoop

启动HDFS，打开终端并切换到Hadoop的sbin目录。 运行start-dfs.sh脚本来启动HDFS。这将依次启动NameNode、Secondary NameNode和所有DataNodes（如果是在集群模式下）。

  cd $HADOOP_HOME/sbin./start-dfs.sh

启动YARN

在同一个sbin目录下，运行start-yarn.sh脚本来启动YARN。这将启动ResourceManager和所有NodeManagers。

./start-yarn.sh

验证Hadoop服务

使用jps命令来验证Hadoop服务是否已经启动。

jps

你应该能看到NameNode, DataNode, SecondaryNameNode, ResourceManager, 和 NodeManager 进程。

停止Hadoop

停止YARN，在Hadoop的sbin目录下运行stop-yarn.sh脚本来停止YARN服务。

./stop-yarn.sh

停止HDFS

接着运行stop-dfs.sh脚本来停止HDFS服务。

./stop-dfs.sh

这将依次停止所有DataNodes、Secondary NameNode和NameNode。
全部停止

如果你想一次性停止所有Hadoop服务，可以运行stop-all.sh脚本。

./stop-all.sh

同样地，如果你想一次性启动所有Hadoop服务，可以运行start-all.sh脚本。

./start-all.sh

请注意，这些脚本可能会根据你的Hadoop版本和配置略有不同。在某些情况下，你可能需要使用完整的路径来引用这些脚本，例如$HADOOP_HOME/sbin/start-dfs.sh。

另外，如果你使用的是Hadoop 2.x或更高版本，YARN的启动和停止脚本名可能与上面提到的不同。例如，在Hadoop 2.x中，YARN的启动脚本可能是start-yarn.sh而不是start-mapred.sh。在使用这些脚本之前，请确保你查阅了你的Hadoop版本的官方文档。

Hadoop集群搭建步骤

搭建Hadoop集群涉及多个步骤，具体取决于你的硬件环境、操作系统选择以及Hadoop的版本。以下是一个基于Linux系统的Hadoop集群搭建的基本流程：

准备阶段

环境规划：确定Hadoop集群的规模，包括Master节点和Slave节点的数量，以及它们之间的网络拓扑结构。
硬件准备：确保所有服务器满足Hadoop的要求，如足够的CPU、内存和磁盘空间。
操作系统安装：在所有服务器上安装相同版本的Linux发行版。
SSH无密码登录：配置Master节点到所有其他节点的SSH无密码登录，便于远程管理。
时钟同步：使用NTP或其他工具同步所有节点的系统时间。
关闭防火墙：暂时关闭防火墙以避免网络问题。

Java环境配置

安装Java：在所有节点上安装Java SE Development Kit (JDK)，并配置JAVA_HOME环境变量。

Hadoop安装与配置

上传Hadoop：将Hadoop的tar.gz压缩包上传至所有节点的指定目录。
解压Hadoop：在每个节点上解压Hadoop压缩包。
配置Hadoop：编辑Hadoop的核心配置文件，如core-site.xml, hdfs-site.xml, mapred-site.xml, yarn-site.xml等，确保它们指向正确的存储目录和网络配置。
环境变量设置：在所有节点上设置HADOOP_HOME环境变量，并更新.bashrc或.profile以包含Hadoop的bin目录。

HDFS格式化与启动服务

格式化HDFS：在Master节点上执行HDFS格式化命令。
启动Hadoop服务：在Master节点上启动HDFS和YARN服务。

测试集群

测试HDFS：上传一些数据到HDFS并检查是否能够正确读取。
测试MapReduce或YARN：运行一个简单的MapReduce任务或Spark作业，确保集群正常工作。

安装额外组件

安装Hive：如果需要SQL-like查询能力，可以在Hadoop集群上安装Hive。
安装HBase：如果需要NoSQL数据库功能，可以安装HBase。

监控与维护：

安装监控工具：例如使用Ganglia或Nagios监控集群健康状态。
定期维护：包括数据平衡、日志分析和软件升级等。

请注意，具体的配置细节会根据Hadoop的版本和你的具体需求有所不同。如果你正在使用Hadoop 2.x或3.x，某些配置文件的名称可能略有不同，比如mapred-site.xml在Hadoop 2.x中被mapreduce-site.xml所取代。

在实际操作中，你可能需要参考Hadoop官方文档和相关的社区指南来获得最新的配置建议和解决潜在的问题

使用Docker搭建集群

使用Docker搭建Hadoop集群可以简化部署过程，同时提供更好的隔离性和可移植性。以下是一个使用Docker和Docker Compose搭建Hadoop伪分布式环境的步骤：
步骤1: 准备Docker镜像

首先，你需要获取Hadoop的Docker镜像。你可以在Docker Hub上找到预构建的Hadoop镜像，或者自己构建一个。这里假设你使用的是bde2020/hadoop-distrubuted镜像。
步骤2: 编写Docker Compose文件

创建一个docker-compose.yml文件，定义你的Hadoop集群。以下是一个示例文件，用于创建一个包含NameNode、DataNode和JobTracker的伪分布式集群：

version: '3'services:namenode:image: bde2020/hadoop-distrubuted:latestcontainer_name: hadoop_namenoderestart: alwaysvolumes:- ./hadoop/namenode:/hadoop/dfs/namenodeenvironment:SERVICE_PRECONDITION: "datanode:9866 jobtracker:9001"ports:- "50070:50070"datanode:image: bde2020/hadoop-distrubuted:latestcontainer_name: hadoop_datanoderestart: alwaysvolumes:- ./hadoop/datanode:/hadoop/dfs/datanodeports:- "9866:9866"jobtracker:image: bde2020/hadoop-distrubuted:latestcontainer_name: hadoop_jobtrackerrestart: alwaysports:- "9001:9001"secondarynamenode:image: bde2020/hadoop-distrubuted:latestcontainer_name: hadoop_secondarynamenoderestart: alwaysvolumes:- ./hadoop/secondarynamenode:/hadoop/dfs/secondarynamenodeports:- "50090:50090"

步骤3: 格式化HDFS并启动集群

在你的Docker Compose文件所在的目录下，运行以下命令来格式化HDFS并启动集群：

# 格式化HDFS
docker-compose run --rm namenode /usr/local/bin/hadoop namenode -format# 启动集群
docker-compose up -d

步骤4: 验证集群

你可以通过访问各个服务的Web UI来验证集群是否正常运行。例如，NameNode的Web UI通常可以通过http://localhost:50070/访问。
步骤5: 使用Hadoop集群

现在，你可以在容器内部或外部运行Hadoop命令，例如上传文件到HDFS或提交MapReduce作业。为了在宿主机上运行Hadoop命令并连接到Docker中的集群，你可能需要配置_/.bashrc或/.profile来包含Docker容器中的Hadoop bin目录。

以上步骤描述了一个基础的Hadoop伪分布式集群的搭建过程。对于全分布式集群，你可能需要在多台物理机器上运行Docker，并调整Docker Compose文件以反映你的网络配置。

使用Hadoop读写数据

在Hadoop中，我们通常使用HDFS（Hadoop Distributed File System）进行数据存储，以及MapReduce或Spark进行数据处理。以下是一个使用Java API进行HDFS数据读写的简单示例：

写入数据到HDFS

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;import java.io.FileOutputStream;
import java.io.OutputStream;
import java.net.URI;public class HDFSWritter {public static void main(String[] args) throws Exception {Configuration conf = new Configuration();FileSystem fs = FileSystem.get(URI.create("hdfs://namenode:9000"), conf, "hdfs");String file = "/path/to/your/file";OutputStream out = fs.create(new Path(file));IOUtils.copyBytes("This is some text", 0, 15, out, conf);IOUtils.closeStream(out);fs.close();}
}

读数据

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;import java.io.InputStream;
import java.net.URI;public class HDFSReader {public static void main(String[] args) throws Exception {Configuration conf = new Configuration();FileSystem fs = FileSystem.get(URI.create("hdfs://namenode:9000"), conf, "hdfs");String file = "/path/to/your/file";InputStream in = fs.open(new Path(file));IOUtils.copyBytes(in, System.out, 4096, false);IOUtils.closeStream(in);fs.close();}
}

注意：在上述代码中，“namenode:9000”应替换为你的Hadoop集群的namenode地址和端口，“hdfs”应替换为你的Hadoop集群的用户名，“/path/to/your/file”应替换为你想要读写文件的实际路径。

此外，你还需要在你的项目中添加Hadoop的依赖，例如在Maven的pom.xml中添加：

<dependencies><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-client</artifactId><version>3.2.1</version></dependency>
</dependencies>

使用Hadoop进行分布式计算

在Hadoop中，主要通过MapReduce框架来进行分布式计算。下面是一个使用Java编写的简单WordCount程序的例子，它展示了如何使用Hadoop MapReduce进行分布式计算。
步骤1: 创建Mapper类

import java.io.IOException;
import java.util.StringTokenizer;import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {private final static 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 tokenizer = new StringTokenizer(line);while (tokenizer.hasMoreTokens()) {word.set(tokenizer.nextToken());context.write(word, one);}}
}

步骤2: 创建Reducer类

import java.io.IOException;import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {private IntWritable result = new IntWritable();public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {int sum = 0;for (IntWritable val : values) {sum += val.get();}result.set(sum);context.write(key, result);}
}

步骤3: 创建Driver类

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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;public class WordCountDriver {public static void main(String[] args) throws Exception {Configuration conf = new Configuration();Job job = Job.getInstance(conf, "word count");job.setJarByClass(WordCountDriver.class);job.setMapperClass(WordCountMapper.class);job.setCombinerClass(WordCountReducer.class);job.setReducerClass(WordCountReducer.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);FileInputFormat.addInputPath(job, new Path(args[0]));FileOutputFormat.setOutputPath(job, new Path(args[1]));System.exit(job.waitForCompletion(true) ? 0 : 1);}
}

运行程序

首先，你需要将输入文本文件放到HDFS上，然后运行你的MapReduce程序。假设你的输入文件在HDFS上的路径是/user/input，输出路径是/user/output，那么你可以在命令行中使用以下命令运行你的程序：

hadoop jar your-jar-file.jar WordCountDriver /user/input /user/output