:大数据行业部署实战3:基于Hadoop的Web版的云盘

简介: :大数据行业部署实战3:基于Hadoop的Web版的云盘

一、实验目的

熟练采用JAVA API访问 HDFS。

二、实验原理

HDFS是hadoop平台的核心组成之一。熟悉使用hadoop平台需要熟练访问HDFS。HDFS的访问方式有多种。可通过web访问,也可通过shell方式或者API方式访问。基本操作有对文件的读、写、追加、删除等。新建文件夹、删除文件夹等。还可显示文件及文件夹的属性。

HDFS主要用到了FileSystem类,相关的接口可以在这里查到:

http://hadoop.apache.org/docs/r2.7.3/api/org/apache/hadoop/fs/FileSystem.html

或通过IDEA,Ctrl+点击 FileSystem类,也可以看到源码。

下面列了FileSystem的常用接口:

三、实验环境

hadoop2.7.3

Java IDE:IDEA

四、实验内容

打开桌面terminal,在家目录下,下载项目,并解压缩

wget http://i9000.net:8888/sgn/HUP/HadoopDeployPro/cloud-disk.zip
unzip cloud-disk.zip

检查是否已启动Hadoop

用jps查看是否启动成功,如果有进程未启动,可以尝试再次启动Hadoop

1.打开IDEA,导入项目“clouddisk”。

选择clouddisk项目的pom.xml

2.修改配置

application.properties文件中修改下面的IP为虚拟机的IP,此时我们写localhost即可

hadoop.namenode.rpc.url=hdfs://localhost:8020

3.修改用户代码观察网盘效果

将如图文件的,将所有的hadoop用户改为ubuntu,涉及创建目录、删除目录等,总共5处需要修改。

4.右键运行

当看到如下截图时:

访问页面:

http://localhost:9090/ (链接到外部网站。)链接到外部网站。

可以看到实验一、实验二在hdfs上的目录,我们还可以自己创建,删除,重命名

无论删除还是增加都会有日志:

NOTE:每一次修改代码都需要重新运行代码,点击如图Rerun图标,重新运行代码,当tomcat与项目代码都启动时,打开浏览器再进行操作

5.开发,实现上传、下载

类名:com.mypro.clouddisk.hdfs.FileSystemImpl

实现upload 、download的代码。

@Override
public void upload(InputStream is, String dstHDFSFile) throws Exception {
//TODO
//代码待实现
System.out.println( "Upload Successfully!" );
}
@Override
public void download(String file, OutputStream os) throws Exception {
//TODO
//代码待实现
System.out.println( "Download Successfully!" );
}

五、代码附录

IndexController

package com.mypro.clouddisk.controller;
import com.mypro.clouddisk.hdfs.IFileSystem;
import com.mypro.clouddisk.model.FileIndex;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Controller;
import org.springframework.ui.Model;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.multipart.MultipartFile;
import org.springframework.web.servlet.mvc.support.RedirectAttributes;
import javax.servlet.http.HttpServletResponse;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.*;
@Controller
public class IndexController {
    @Autowired
    private IFileSystem fileSystem = null;
    @RequestMapping("/")
    public String index(String path,Model model) {
        FileIndex fileIndex = new FileIndex();
        path = (path==null || path.trim().isEmpty()) ? "/": path.trim();
        fileIndex.setPath(path);
        String fileName = fileSystem.getFileName(path);
        fileIndex.setName(fileName);
        model.addAttribute("rootDir",fileIndex);
        List<FileIndex> list = new ArrayList<FileIndex>();
        try {
            list = fileSystem.ls(path);
        } catch (Exception e) {
            e.printStackTrace();
        }
        model.addAttribute("rootFiles",list);
        return "index";
    }
    @GetMapping("/download")
    public String download(HttpServletResponse response, @RequestParam String file) throws Exception {
//        String filename="xxx.txt";
        File hFile = new File(file);
        String filename = hFile.getName();
        response.setHeader("Content-Disposition", "attachment;fileName=" + filename);
        fileSystem.download(file,response.getOutputStream());
        return null;
    }
    /**
     *
     * @param path  //???????
     * @return
     * @throws Exception
     */
    @RequestMapping("/delete")
    public String delete(@RequestParam String path) throws Exception {
        String parentPath = fileSystem.rm(path);
        //??????
        return "redirect:./?path=" + parentPath;
    }
    @ResponseBody
    @RequestMapping(value="/checkMD5",method=RequestMethod.POST)
    public String checkMD5(String md5code){
        //??????????MD5??md5code???
        return "no";
    }
    @PostMapping("/uploadFile")
    public String singleFileUpload(@RequestParam("file") MultipartFile file,
                                   String parentPath,RedirectAttributes redirectAttributes) throws IOException {
        String fileName = getOriginalFilename(file.getOriginalFilename());
        InputStream is = file.getInputStream();
        Long size = file.getSize();
        if (file.isEmpty()) {
            redirectAttributes.addFlashAttribute("message", "Please select a file to upload");
            return "redirect:uploadStatus";
        }
        try {
            String dstPath = parentPath.endsWith("/")? (parentPath+fileName) : (parentPath +"/"+fileName);
            fileSystem.upload(is,dstPath);
            redirectAttributes.addFlashAttribute("message",
                    "You successfully uploaded '" + file.getOriginalFilename() + "'");
        } catch (IOException e) {
            e.printStackTrace();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return "redirect:./?path="+parentPath;
    }
    @PostMapping("/mkdir")
    public String mkdir(String directName,String parentPath) throws Exception {
        String newDir = parentPath.endsWith("/")?(parentPath+directName):(parentPath+"/"+directName);
        fileSystem.mkdir(newDir);
        return "redirect:./?path="+parentPath;
    }
    @RequestMapping("renameForm")
    public String renameForm(String directName,String isRoot,String path) throws Exception {
        //??fileindex?name?path
        //???????path
        String[] arr= fileSystem.rename(path,directName);
        //?????????????????????????????????????????????????????????
        if(isRoot!=null && isRoot.equals("yes")){
            return "redirect:./?path=" + arr[1];  //mypath
        }else{
            return "redirect:./?path=" + arr[0];//parent
        }
    }
    @RequestMapping("/getOptionalPath")
    @ResponseBody
    public List getOptionalPath(String path){
        //?????fileIndexId?????????????????????????????
        List result =  fileSystem.getOptionTranPath(path);
        return result;
    }
    @RequestMapping("/searchFiles")
    public String searchFiles(String keyWord,@RequestParam(value="pageNum",defaultValue="1")int pageNum,Model model) throws Exception {
        int pageSize = 3;
        List<FileIndex> result = fileSystem.searchFileByPage(keyWord,pageSize,pageNum);
        com.github.pagehelper.Page<FileIndex> page =new com.github.pagehelper.Page<FileIndex>();
        page.addAll(result);
        model.addAttribute("result", page);
        model.addAttribute("keyWord",keyWord);
        return "searchResult";
    }
    @RequestMapping("/stasticFiles")
    public String stasticFiles(Model model){
//        List staticResult = fileSystem.getStaticNums();
        List<Map> staticResult = new ArrayList<Map>();
        Map map = new HashMap();
        map.put("doc_number",100);
        map.put("video_number",87);
        map.put("pic_number",66);
        map.put("code_number",44);
        map.put("other_number",23);
        staticResult.add(map);
        model.addAttribute("staticResult", staticResult);
        return "stasticfiles";
    }
    private String getOriginalFilename(String originalFilename){
        if(originalFilename == null)
        {
            return "";
        }
        if(originalFilename.contains("/") || originalFilename.contains("\\")){
            File file = new File(originalFilename);
            return file.getName();
        }
        return  originalFilename;
    }
}
}

FileSystemImpl

package com.mypro.clouddisk.hdfs;
import com.mypro.clouddisk.model.FileIndex;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import java.io.*;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
@Component
public class FileSystemImpl implements IFileSystem {
    Logger logger = LoggerFactory.getLogger(FileSystemImpl.class);
    @Value("${hadoop.namenode.rpc.url}")
    private String namenodeRpcUrl;
//    private static String NAMENODE_RPC="hdfs://192.168.72.128:8020";
    @Override
    public List<FileIndex> ls(String dir) throws Exception{
        Configuration conf=new Configuration();
        //??NameNode??
        URI uri=new URI(namenodeRpcUrl);
        //?????,??FileSystem??
        FileSystem fs=FileSystem.get(uri,conf,"ubuntu");
        Path path = new Path(dir);
        //????????
        if(! fs.exists(path)){
            logger.error("dir:"+dir+" not exists!");
            throw new RuntimeException("dir:"+dir+" not exists!");
        }
        List<FileIndex> list = new ArrayList<FileIndex>();
        FileStatus[] filesStatus = fs.listStatus(path);
        for(FileStatus f:filesStatus){
            FileIndex fileIndex = new FileIndex();
            fileIndex.setIsFile(f.isDirectory()?"0":"1");
            fileIndex.setName(f.getPath().getName());
            fileIndex.setPath(f.getPath().toUri().getPath());
            fileIndex.setCreateTime(new Date());
            list.add(fileIndex);
        }
        //??????FileSystem????
        fs.close();
        return list;
    }
    @Override
    public void mkdir(String dir) throws Exception{
        Configuration conf=new Configuration();
        //??NameNode??
        URI uri=new URI(namenodeRpcUrl);
        //?????,??FileSystem??
        FileSystem fs=FileSystem.get(uri,conf,"ubuntu");
        fs.mkdirs(new Path(dir));
        //??????FileSystem????client
        fs.close();
        System.out.println( "mkdir "+dir+" Successfully!" );
    }
    @Override
    /**
     * ???????
     */
    public String rm(String path) throws Exception {
        Configuration conf=new Configuration();
        //??NameNode??
        URI uri=new URI(namenodeRpcUrl);
        //?????,??FileSystem??
        FileSystem fs=FileSystem.get(uri,conf,"ubuntu");
        Path filePath = new Path(path);
        fs.delete(filePath,true);
        //??????FileSystem????client
        fs.close();
        System.out.println( "Delete "+path+" Successfully!" );
        return filePath.getParent().toUri().getPath();
    }
    @Override
    public void upload(InputStream is, String dstHDFSFile) throws Exception {
        //TODO
    //???
        System.out.println( "Upload Successfully!" );
    }
    @Override
    public void download(String file, OutputStream os) throws Exception {
            //TODO
    //???
        System.out.println( "Download Successfully!" );
    }
    @Override
    public void mv() {
    }
    @Override
    public String[] rename(String path, String dirName) throws Exception {
        Configuration conf=new Configuration();
        //??NameNode??
        URI uri=new URI(namenodeRpcUrl);
        //?????,??FileSystem??
        FileSystem fs=FileSystem.get(uri,conf,"ubuntu");
        Path oldPath = new Path(path);
        if(oldPath.getParent() ==null){
            String[] arr = new String[2];
            arr[0]="/";
            arr[1]="/";
            return arr;
        }
        String parentPath = oldPath.getParent().toUri().getPath();
        String newPathStr = parentPath.endsWith("/")?(parentPath+dirName):(parentPath + "/" +dirName);
        Path newPath = new Path(newPathStr );
        fs.rename(oldPath,newPath);
        //??????FileSystem????client
        fs.close();
        String[] arr = new String[2];
        arr[0]=parentPath;
        arr[1]=newPathStr;
        System.out.println( "rename Successfully!" );
        return arr;
    }
    @Override
    public String getFileName(String path) {
        Path filePath = new Path(path);
        return filePath.getName();
    }
    @Override
    public List getOptionTranPath(String path) {
        return new ArrayList();
    }
    @Override
    public List<FileIndex> searchFileByPage(String keyWord, int pageSize, int pageNum) throws Exception{
        Configuration conf=new Configuration();
        //??NameNode??
        URI uri=new URI(namenodeRpcUrl);
        //?????,??FileSystem??
        FileSystem fs=FileSystem.get(uri,conf,"ubuntu");
        ArrayList<FileStatus> results = null;
        PathFilter filter = new PathFilter() {
            @Override
            public boolean accept(Path path) {
                if(keyWord == null || keyWord.trim().isEmpty()){
                    return false;
                }
                if(path.getName().contains(keyWord)){
                    return  true;
                }
                return false;
            }
        };
        List<FileIndex> list = new ArrayList<FileIndex>();
        FileStatus[] fileStatusArr = fs.listStatus(new Path("/"),filter);
        for(FileStatus status :fileStatusArr){
            FileIndex fileIndex = new FileIndex();
            fileIndex.setPath(status.getPath().toUri().getPath());
            fileIndex.setName(status.getPath().getName());
            fileIndex.setIsFile(status.isFile()?"1":"0");
            list.add(fileIndex);
        }
        //??????FileSystem????client
        fs.close();
        System.out.println( "Search Successfully!" );
        return list;
    }
    @Override
    public List getStaticNums() {
        return null;
    }
}

FileSystem

package com.mypro.clouddisk.hdfs;
import com.mypro.clouddisk.model.FileIndex;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.URISyntaxException;
import java.util.List;
public interface IFileSystem {
    /**
     * ls ????????
     * @return
     */
    List<FileIndex> ls(String dir) throws Exception;
    /**
     * ????
     * @return
     */
    void mkdir(String dir)throws Exception;
    /**
     * ???????
     * ??????
     */
    String rm(String path) throws Exception;
    /**
     * ??
     */
    void upload(InputStream is, String dstHDFSFile) throws Exception;
    /**
     * ??
     */
    void download(String file, OutputStream os) throws Exception;
    /**
     * ???
     */
    void mv();
    /**
     * ???
     */
    String[] rename(String path,String dirName) throws Exception;
    /**
     * ?????
     * @param path
     * @return
     */
    String getFileName(String path);
    //?????fileIndexId?????????????????????????????
    List getOptionTranPath(String path);
    List<FileIndex> searchFileByPage(String keyWord, int pageSize, int pageNum) throws Exception;
    List getStaticNums();
}


相关实践学习
简单用户画像分析
本场景主要介绍基于海量日志数据进行简单用户画像分析为背景,如何通过使用DataWorks完成数据采集 、加工数据、配置数据质量监控和数据可视化展现等任务。
SaaS 模式云数据仓库必修课
本课程由阿里云开发者社区和阿里云大数据团队共同出品,是SaaS模式云原生数据仓库领导者MaxCompute核心课程。本课程由阿里云资深产品和技术专家们从概念到方法,从场景到实践,体系化的将阿里巴巴飞天大数据平台10多年的经过验证的方法与实践深入浅出的讲给开发者们。帮助大数据开发者快速了解并掌握SaaS模式的云原生的数据仓库,助力开发者学习了解先进的技术栈,并能在实际业务中敏捷的进行大数据分析,赋能企业业务。 通过本课程可以了解SaaS模式云原生数据仓库领导者MaxCompute核心功能及典型适用场景,可应用MaxCompute实现数仓搭建,快速进行大数据分析。适合大数据工程师、大数据分析师 大量数据需要处理、存储和管理,需要搭建数据仓库?学它! 没有足够人员和经验来运维大数据平台,不想自建IDC买机器,需要免运维的大数据平台?会SQL就等于会大数据?学它! 想知道大数据用得对不对,想用更少的钱得到持续演进的数仓能力?获得极致弹性的计算资源和更好的性能,以及持续保护数据安全的生产环境?学它! 想要获得灵活的分析能力,快速洞察数据规律特征?想要兼得数据湖的灵活性与数据仓库的成长性?学它! 出品人:阿里云大数据产品及研发团队专家 产品 MaxCompute 官网 https://www.aliyun.com/product/odps&nbsp;
目录
相关文章
|
3天前
|
前端开发 JavaScript BI
Django教程第5章 | Web开发实战-数据统计图表(echarts、highchart)
使用echarts和highcharts图表库实现折线图、柱状图、饼图和数据集图
24 2
|
1月前
|
分布式计算 DataWorks MaxCompute
DataWorks中odps到容器部署starrocks的单表同步遇到写入问题
【1月更文挑战第6天】【1月更文挑战第29篇】DataWorks中odps到容器部署starrocks的单表同步遇到写入问题
23 3
|
2月前
|
SQL 分布式计算 数据可视化
滴滴出行大数据数仓实战
滴滴出行大数据数仓实战
59 0
滴滴出行大数据数仓实战
|
27天前
|
前端开发 JavaScript 测试技术
深入实战:构建现代化的Web前端应用
深入实战:构建现代化的Web前端应用
21 0
|
3天前
|
存储 数据库 Python
Django教程第6章 | web开发实战-文件上传(导入文件、上传图片)
web应用实战:导入文件解析到DB,上传图片
34 0
|
3天前
|
存储 前端开发 JavaScript
Django教程第4章 | Web开发实战-三种验证码实现
手动生成验证码,自动生成验证码,滑动验证码。
28 0
|
3天前
|
存储 中间件 数据安全/隐私保护
Django教程第3章 | Web开发实战-登录
登录案例、Djiango中间件
24 2
|
3天前
|
JavaScript 关系型数据库 MySQL
Django教程第2章| Web开发实战-用户管理
基于Django实现用户管理:增删改查,搜索,分页。
23 0
|
23天前
|
缓存 安全 网络安全
【网络安全】Web缓存欺骗攻击原理及攻防实战
【网络安全】Web缓存欺骗攻击原理及攻防实战
23 0
【网络安全】Web缓存欺骗攻击原理及攻防实战
|
24天前
|
分布式计算 大数据 Java
Spark 大数据实战:基于 RDD 的大数据处理分析
Spark 大数据实战:基于 RDD 的大数据处理分析
52 0