Hadoop 学习笔记

环境配置

使用Docker部署Hadoop

拉取一个CentOS镜像作为Hadoop运行

docker pull centos:8

配置Java与SSH环境

创建容器

创建一个名为 java_ssh_proto的容器

docker run -d --name=java_ssh_proto --privileged centos:8 /usr/sbin/init

进入容器

docker exec -it java_ssh_proto bash

配置镜像

下载阿里云镜像配置

curl -o /etc/yum.repos.d/CentOS-Base.repo http://mirrors.aliyun.com/repo/Centos-8.repo

修改配置文件

sed -i -e '/mirrors.cloud.aliyuncs.com/d' -e '/mirrors.aliyuncs.com/d' /etc/yum.repos.d/CentOS-Base.repo
sed -i 's/releasever\//releasever-stream\//g' /etc/yum.repos.d/CentOS-Base.repo

重置缓存

yum clean all 
yum makecache

如果出现以下错误

解决方法：

进入repos目录

cd /etc/yum.repos.d/

修改文件内容

sed -i 's/mirrorlist/#mirrorlist/g' /etc/yum.repos.d/CentOS-*
sed -i 's|#baseurl=http://mirror.centos.org|baseurl=http://vault.centos.org|g' /etc/yum.repos.d/CentOS-*

再次缓存更新即可解决

yum makecache

稍等片刻即可

安装OpenJDK8和SSH服务

yum install -y java-1.8.0-openjdk-devel openssh-clients openssh-server

启动SSH服务

systemctl enable sshd && systemctl start sshd

至此，包含 Java 与SSH环境的容器创建完毕

使用ctrl+d或者exit命令退出容器

停止容器并保存

docker stop java_ssh_proto
docker commit java_ssh_proto java_ssh

安装 Hadoop

创建 Hadoop 单机容器

用之前保存的 java_ssh 镜像创建容器 hadoop_single

docker run -d --name=hadoop_single --privileged java_ssh /usr/sbin/init

使用阿里云镜像下载Hadoop

wget https://mirrors.aliyun.com/apache/hadoop/common/hadoop-3.2.3/hadoop-3.2.3.tar.gz

镜像下载速度也勉勉强强。。。但是至少是比官网快多了

将下载好的 hadoop 压缩包拷贝到容器中的 /root 目录下

docker cp hadoop-3.2.3.tar.gz hadoop_single:/root/

进入容器

docker exec -it hadoop_single bash

进入 /root 目录

cd /root

解压压缩包

tar -zxf hadoop-3.2.3.tar.gz

解压后将得到一个文件夹 hadoop-3.2.3，现在把它拷贝到一个常用的地方：

mv hadoop-3.2.3 /usr/local/hadoop

配置环境变量

echo "export HADOOP_HOME=/usr/local/hadoop" >> /etc/bashrc
echo "export PATH=$PATH:$HADOOP_HOME/bin:$HADOOP_HOME/sbin" >> /etc/bashrc

退出容器并重新进入，输入echo $HADOOP_HOME 的结果应该是 /usr/local/hadoop

配置hadoop内置的环境变量

echo "export JAVA_HOME=/usr" >> $HADOOP_HOME/etc/hadoop/hadoop-env.sh
echo "export HADOOP_HOME=/usr/local/hadoop" >> $HADOOP_HOME/etc/hadoop/hadoop-env.sh

输入下列命令查看是否成功

hadoop version

（如果报 command not found 便再执行一次添加环境变量步骤并退出重进容器）

至此，Hadoop容器制作完毕，接下来进入实际使用学习。

HDFS （Hadoop Distributed File System）

分布式存储系统

支持海量数据的存储，成百上千的计算机组成存储集群，HDFS可以运行在低成本的硬件之上，具有的高容错、高可靠性、高可扩展性、高吞吐率等特征，非常适合大规模数据集上的应用。

优点

• 高容错性
• 适合批处理
• 适合大数据处理
• 流式文件访问
• 可构建在廉价机器上

缺点

• 不适合低延迟数据访问
• 不适合小文件存取
• 不适合并发写入、文件随机修改

HDFS操作

命令操作HDFS

# 显示目录 / 下的文件
hdfs dfs -ls /
# 新建文件夹，绝对路径
hdfs dfs -mkdir /test
# 上传文件
hdfs dfs -put test.txt /test/
# 下载文件
hdfs dfs -get /test/test.txt
# 输出文件内容
hdfs dfs -cat /test/test.txt

Web端操作HDFS

打开http://192.168.9.200:9870/，可以对HDFS进行操作

文件在这里管理

可视化操作还是简单一些

Java操作HDFS

1. 添加依赖

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

2. 建立连接

public void setUp()throws Exception{
  System.out.println("开始建立与HDFS的连接");
  configuration=new Configuration();
  fileSystem=FileSystem.get(new URI(HDFS_PATH),configuration,"hadoop");
  }

3. 文件操作

    Configuration configuration=null;
  FileSystem fileSystem=null;
public static final String HDFS_PATH="hdfs://192.168.9.200:9000";

/**
 * 在 hdfs中新建文件夹
 *
 * @throws Exception
 */
@Test public void mkdir()throws Exception{
  fileSystem.mkdirs(new Path("/JavaDemo/test"));
  }

/**
 * 创建文件
 *
 * @throws Exception
 */

@Test public void create()throws Exception{
  FSDataOutputStream outputStream=fileSystem.create(new Path("/JavaDemo/test/haha.txt"));
  outputStream.write("hello bigdata from javaDemo".getBytes());
  outputStream.flush();
  outputStream.close();
  }


/**
 * 查看文件 hdfs -fs -cat file
 *
 * @throws Exception
 */
@Test public void cat()throws Exception{
  FSDataInputStream in=fileSystem.open(new Path("/JavaDemo/test/haha.txt"));
  IOUtils.copyBytes(in,System.out,1024);
  in.close();
  }

  /**
   * 重命名文件
   *
   * @throws Exception
   */
  @Test public void rename()throws Exception{
  Path oldPath=new Path("/JavaDemo/test/haha.txt");
  Path newPath=new Path("/JavaDemo/test/hehe.txt");
  fileSystem.rename(oldPath,newPath);
  }

  /**
   * 上传文件到HDFS
   *
   * @throws Exception
   */
  @Test public void copyFromLocalFile()throws Exception{
  Path loacalPath=new Path("hello.txt");
  Path hdfsPath=new Path("/");
  fileSystem.copyFromLocalFile(loacalPath,hdfsPath);
  }


  /**
   * 上传文件到HDFS带进度信息
   *
   * @throws Exception
   */

  @Test public void copyFromLocalFileWithProgress()throws Exception{
  InputStream in=new BufferedInputStream(Files.newInputStream(new File("hbase-2.2.7-bin.tar.gz").toPath()));
  FSDataOutputStream ouput=fileSystem.create(new Path("/JavaDemo/test/hbase-2.2.7-bin.tar.gz"),()->{
  System.out.print(".");
  });
  IOUtils.copyBytes(in,ouput,4096);
  }

  /**
   * 下载文件到HDFS
   *
   * @throws Exception
   */
  @Test public void copyToLocalFile()throws Exception{
  Path hdfsPath=new Path("/JavaDemo/test/haha.txt");
  Path loacalPath=new Path("./haha.txt");
  //  useRawLocalFileSystem
  fileSystem.copyToLocalFile(false,hdfsPath,loacalPath,true);
  }

  /**
   * 查看某个目录下所有文件
   *
   * @throws Exception
   */
  @Test public void listFiles()throws Exception{
  FileStatus[]fileStatuses=fileSystem.listStatus(new Path("/"));
  for(FileStatus f:fileStatuses){
  String isDir=f.isDirectory()?"文件夹":"文件";
  short replication=f.getReplication();
  long len=f.getLen();
  String path=f.getPath().toString();

  System.out.println(isDir+"\t"+replication+"\t"+len+"\t"+path);

  }
  }

  /**
   * 删除文件
   *
   * @throws Exception
   */
  @Test public void delete()throws IOException{
  fileSystem.delete(new Path("/JavaDemo/haha.txt"),true);
  }

4. 关闭连接

public void tearDown(){
  configuration=null;
  fileSystem=null;
  System.out.println("关闭与HDFS的连接");
  }

完整测试文件

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

import java.io.*;
import java.net.URI;
import java.nio.file.Files;

/**
 * @author Gettler
 * Java 操作HDFS
 */
public class HdfsDemo {
  Configuration configuration = null;
  FileSystem fileSystem = null;
  public static final String HDFS_PATH = "hdfs://192.168.9.200:9000";

  /**
   * 在 hdfs中新建文件夹
   *
   * @throws Exception
   */
  @Test
  public void mkdir() throws Exception {
    fileSystem.mkdirs(new Path("/JavaDemo/test"));
  }

  /**
   * 创建文件
   *
   * @throws Exception
   */

  @Test
  public void create() throws Exception {
    FSDataOutputStream outputStream = fileSystem.create(new Path("/JavaDemo/test/haha.txt"));
    outputStream.write("hello bigdata from javaDemo".getBytes());
    outputStream.flush();
    outputStream.close();
  }


  /**
   * 查看文件 hdfs -fs -cat file
   *
   * @throws Exception
   */
  @Test
  public void cat() throws Exception {
    FSDataInputStream in = fileSystem.open(new Path("/JavaDemo/test/haha.txt"));
    IOUtils.copyBytes(in, System.out, 1024);
    in.close();
  }

  /**
   * 重命名文件
   *
   * @throws Exception
   */
  @Test
  public void rename() throws Exception {
    Path oldPath = new Path("/JavaDemo/test/haha.txt");
    Path newPath = new Path("/JavaDemo/test/hehe.txt");
    fileSystem.rename(oldPath, newPath);
  }

  /**
   * 上传文件到HDFS
   *
   * @throws Exception
   */
  @Test
  public void copyFromLocalFile() throws Exception {
    Path loacalPath = new Path("hello.txt");
    Path hdfsPath = new Path("/");
    fileSystem.copyFromLocalFile(loacalPath, hdfsPath);
  }


  /**
   * 上传文件到HDFS带进度信息
   *
   * @throws Exception
   */

  @Test
  public void copyFromLocalFileWithProgress() throws Exception {
    InputStream in = new BufferedInputStream(Files.newInputStream(new File("hbase-2.2.7-bin.tar.gz").toPath()));
    FSDataOutputStream ouput = fileSystem.create(new Path("/JavaDemo/test/hbase-2.2.7-bin.tar.gz"), () -> {
      System.out.print(".");
    });
    IOUtils.copyBytes(in, ouput, 4096);
  }

  /**
   * 下载文件到HDFS
   *
   * @throws Exception
   */
  @Test
  public void copyToLocalFile() throws Exception {
    Path hdfsPath = new Path("/JavaDemo/test/haha.txt");
    Path loacalPath = new Path("./haha.txt");
    //  useRawLocalFileSystem
    fileSystem.copyToLocalFile(false, hdfsPath, loacalPath, true);
  }

  /**
   * 查看某个目录下所有文件
   *
   * @throws Exception
   */
  @Test
  public void listFiles() throws Exception {
    FileStatus[] fileStatuses = fileSystem.listStatus(new Path("/"));
    for (FileStatus f : fileStatuses) {
      String isDir = f.isDirectory() ? "文件夹" : "文件";
      short replication = f.getReplication();
      long len = f.getLen();
      String path = f.getPath().toString();

      System.out.println(isDir + "\t" + replication + "\t" + len + "\t" + path);

    }
  }

  /**
   * 删除文件
   *
   * @throws Exception
   */
  @Test
  public void delete() throws IOException {
    fileSystem.delete(new Path("/JavaDemo/haha.txt"), true);
  }


  //测试之前执行的代码
  @Before
  public void setUp() throws Exception {
    System.out.println("开始建立与HDFS的连接");
    configuration = new Configuration();
    fileSystem = FileSystem.get(new URI(HDFS_PATH), configuration, "hadoop");
  }

  //测试之完执行的代码
  @After
  public void tearDown() {
    configuration = null;
    fileSystem = null;
    System.out.println("关闭与HDFS的连接");
  }
}