/**

 * package: com.cloudera.hbase

 * describe: 使用Java将图片生成sequence file并保存到HBase

 * creat_user: Fayson

 * email: htechinfo@163.com

 * creat_date: 2017/11/30

 * creat_time: 上午12:49

 * 公众号：Hadoop实操

 */

public class SequenceFileTest {

    //HDFS路径

    static String inpath = "/fayson/picHbase";

    static String outpath = "/fayson/out";

    static SequenceFile.Writer writer = null;

    static HTable htable = null;

    public static void main(String[] args) throws Exception{

        //inpath = args[0];

        //outpath = args[1];

        //String zklist = args[2];

        //HBase入库

        Configuration hbaseConf = HBaseConfiguration.create();

        hbaseConf.set("hbase.zookeeper.property.clientPort", "2181");

        hbaseConf.setStrings("hbase.zookeeper.quorum", "ip-172-31-5-38.ap-southeast-1.compute.internal");

        //指定表名

        htable = new HTable(hbaseConf,"picHbase");

        //设置读取本地磁盘文件

        Configuration conf = new Configuration();

        //conf.addResource(new Path("C:\\Users\\17534\\eclipse-workspace\\hbaseexmaple\\core-site.xml"));

        //conf.addResource(new Path("C:\\Users\\17534\\eclipse-workspace\\hbaseexmaple\\hdfs-site.xml"));

      

        URI uri = new URI(inpath);

        FileSystem fileSystem = FileSystem.get(uri, conf,"hdfs");

        //实例化writer对象

        writer = SequenceFile.createWriter(fileSystem, conf, new Path(outpath), Text.class, BytesWritable.class);

        //递归遍历文件夹，并将文件下的文件写入sequenceFile文件

        listFileAndWriteToSequenceFile(fileSystem,inpath);

        

        //关闭流

        org.apache.hadoop.io.IOUtils.closeStream(writer);

        //读取所有文件

        URI seqURI = new URI(outpath);

        FileSystem fileSystemSeq = FileSystem.get(seqURI, conf);

        SequenceFile.Reader reader = new SequenceFile.Reader(fileSystemSeq, new Path(outpath), conf);

        Text key = new Text();

        BytesWritable val = new BytesWritable();

//        key = (Text) ReflectionUtils.newInstance(reader.getKeyClass(), conf);

//        val = (BytesWritable) ReflectionUtils.newInstance(reader.getValueClass(), conf);

        int i = 0;

        while(reader.next(key, val)){

            String temp = key.toString();

            temp = temp.substring(temp.lastIndexOf("/") + 1);

//            temp = temp.substring(temp.indexOf("Image")+6, temp.indexOf("."));

//            String[] tmp = temp.split("/");

            //rowKey 设计

            String rowKey = temp;

//            String rowKey = Integer.valueOf(tmp[0])-1+"_"+Integer.valueOf(tmp[1])/2+"_"+Integer.valueOf(tmp[2])/2;

            System.out.println(rowKey);

            //指定ROWKEY的值

            Put put = new Put(Bytes.toBytes(rowKey));

            //指定列簇名称、列修饰符、列值 temp.getBytes()

            put.addColumn("picinfo".getBytes(), "content".getBytes() , val.getBytes());

            htable.put(put);

        }

        htable.close();

        org.apache.hadoop.io.IOUtils.closeStream(reader);

    }

    /****

     * 递归文件;并将文件写成SequenceFile文件

     * @param fileSystem

     * @param path

     * @throws Exception

     */

    public static void listFileAndWriteToSequenceFile(FileSystem fileSystem,String path) throws Exception{

        final FileStatus[] listStatuses = fileSystem.listStatus(new Path(path));

        for (FileStatus fileStatus : listStatuses) {

            if(fileStatus.isFile()){

                Text fileText = new Text(fileStatus.getPath().toString());

                System.out.println(fileText.toString());

                //返回一个SequenceFile.Writer实例 需要数据流和path对象 将数据写入了path对象

                FSDataInputStream in = fileSystem.open(new Path(fileText.toString()));

                byte[] buffer = IOUtils.toByteArray(in);

                in.read(buffer);

                BytesWritable value = new BytesWritable(buffer);

                //写成SequenceFile文件

                writer.append(fileText, value);

            }

            if(fileStatus.isDirectory()){

                listFileAndWriteToSequenceFile(fileSystem,fileStatus.getPath().toString());

            }

        }

    }

}

-----上传文件到hdfs

package com.bigdata.log.controller;

import java.io.IOException;

import java.io.InputStream;

import java.net.URI;

import javax.servlet.http.HttpServletRequest;

import javax.servlet.http.HttpServletResponse;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.Path;

import org.springframework.stereotype.Controller;

import org.springframework.web.bind.annotation.RequestMapping;

import org.springframework.web.bind.annotation.RequestMethod;

import org.springframework.web.bind.annotation.RequestParam;

import org.springframework.web.bind.annotation.ResponseBody;

import org.springframework.web.multipart.MultipartFile;

@Controller

public class Local2HdfsController {

/**

* 单个文件

* @param file

* @param request

* @param response

* @return

* @throws Exception

*/

@RequestMapping(value = "/upload2hdfs", method = {RequestMethod.POST, RequestMethod.GET})

@ResponseBody

    public String upload(@RequestParam("file") MultipartFile file, HttpServletRequest request,HttpServletResponse response) throws Exception {

        HDFSClientUtil hdfsClient = new HDFSClientUtil();

        InputStream in = file.getInputStream();

        

        //上传文件到/input/upload/下

        String tempPathString = "/input/upload/";

        //文件名

        String name = file.getOriginalFilename();

        

        String flag = "";

        

        if(!name.equals("")) {

       

        //上传完成后的路径

            String cloudPath = tempPathString + name;

            try {

            hdfsClient.uploadFile(in, cloudPath);

            flag =  "ok";

    } catch (Exception e) {

    System.out.println("上传失败");

    flag =  "fail";

    }

        }

        

        in.close();

        return flag;

        

    }

/**

* 多文件批量上传

* @param file

* @param request

* @param response

* @return

* @throws Exception

*/

@RequestMapping(value = "/upload2hdfsMore", method = {RequestMethod.POST, RequestMethod.GET})

@ResponseBody

    public String uploads(@RequestParam("file") MultipartFile[] file, HttpServletRequest request,HttpServletResponse response) throws Exception {

        

        HDFSClientUtil hdfsClient = new HDFSClientUtil();

        

        //上传文件到/input/upload/下

        //String tempPathString = "/input/upload/";

        String tempPathString = "/tmp/picHbase/";

       

        String flag = "";

        for (int i = 0; i < file.length; i++) {

            InputStream in = file[i].getInputStream();

           

            //文件名

            String name = file[i].getOriginalFilename();

            

            //上传完成后的路径

            String cloudPath = tempPathString + name;

            

            if (!name.equals("")) {

            try {

                    hdfsClient.uploadFile(in, cloudPath);

                    flag =  "ok";

} catch (Exception e) {

System.out.println("上传失败");

    flag =  "fail";

}

           

                in.close();

}

               

        }

        return flag;

    }

 

 

 

 

/**

     * @function 本地文件上传至 HDFS

     * @param source 原文件路径

     * @param dest  目的文件路径

     * @throws IOException

     * @throws URISyntaxException；

     * 

     * 直接读取本地文件夹，将文件夹中的文件上传至hdfs，暂时不使用该方法

     */

@RequestMapping(value = "/uploadTest.do", method = RequestMethod.POST)

@ResponseBody

    public void copyFromLocal(HttpServletRequest request,HttpServletResponse response) throws Exception {

//String source = request.getParameter("source")==null?"":request.getParameter("source");

    String source = "D://tmp";

        //String source = "./data/weibo.txt";

        // hdfs文件路径

        String dest = "hdfs://199.66.68.112:8020/manyPic/";

        // 读取hadoop文件系统的配置

        Configuration conf = new Configuration();

        URI uri = new URI("hdfs://199.66.68.112:8020");

        // FileSystem是用户操作HDFS的核心类，它获得URI对应的HDFS文件系统

        FileSystem fileSystem = FileSystem.get(uri, conf,"hdfs");

        // 源文件路径

        Path srcPath = new Path(source);

        // 目的路径

        Path dstPath = new Path(dest);

        // 查看目的路径是否存在

        if (!(fileSystem.exists(dstPath))) {

            // 如果路径不存在，即刻创建

            fileSystem.mkdirs(dstPath);

        }

        // 得到本地文件名称

        String filename = source.substring(source.lastIndexOf('/') + 1,source.length());

        try {

            // 将本地文件上传到HDFS

            fileSystem.copyFromLocalFile(srcPath, dstPath);

            

            System.out.println("File " + filename + " copied to " + dest);

        } catch (Exception e) {

            System.err.println("Exception caught! :" + e);

            System.exit(1);

        } finally {

            fileSystem.close();

        }

    }

 

}

---显示图片和视频到jsp页面中

package com.bigdata.log.controller;

import java.io.File;

import java.io.FileOutputStream;

import java.io.IOException;

import java.io.OutputStream;

import java.net.URI;

import java.util.ArrayList;

import javax.servlet.http.HttpServletRequest;

import javax.servlet.http.HttpServletResponse;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FSDataInputStream;

import org.apache.hadoop.fs.FileStatus;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.hbase.HBaseConfiguration;

import org.apache.hadoop.hbase.client.Get;

import org.apache.hadoop.hbase.client.HTable;

import org.apache.hadoop.hbase.client.Result;

import org.apache.hadoop.hbase.util.Bytes;

import org.apache.hadoop.io.IOUtils;

import org.springframework.stereotype.Controller;

import org.springframework.web.bind.annotation.RequestMapping;

import org.springframework.web.bind.annotation.RequestMethod;

import org.springframework.web.bind.annotation.ResponseBody;

@Controller

public class HdfsVideoController {

/**

* 视频播放

* @param request

* @param response

* @throws IllegalArgumentException

* @throws IOException

*/

@RequestMapping(value = "/playVideo.do", method = {RequestMethod.POST, RequestMethod.GET})

@ResponseBody

public void test(HttpServletRequest request, HttpServletResponse response) throws IllegalArgumentException, IOException{

String fpath=request.getParameter("fpath")==null?"":request.getParameter("fpath");

    if(fpath==null)

   return;

    

    //199.66.68.111:8020/.....

String filename=fpath;

//创建configuration对象conf

        Configuration conf = new Configuration();

        //指定hdfs的nameservice为cluster1,是NameNode的URI-Uniform Resource Locator

        conf.set("fs.defaultFS","hdfs://ns");

        //指定hdfs的nameservice为cluster1

        conf.set("dfs.nameservices","ns");

        //cluster1下面有两个NameNode，分别是nna，nns active/standby

        conf.set("dfs.ha.namenodes.ns","nna,nns");

        //nna的RPC通信地址

        conf.set("dfs.namenode.rpc-address.ns.nna","199.66.68.111:8020");

        //nns的RPC通信地址

        conf.set("dfs.namenode.rpc-address.ns.nns","199.66.68.112:8020");

        //配置失败自动切换实现方式

        conf.set("dfs.client.failover.proxy.provider.ns", "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider");

FileSystem fs = null; 

FSDataInputStream in=null;

try {

fs = FileSystem.get(URI.create(filename),conf);

in=fs.open(new Path(filename));

} catch (IOException e) {

e.printStackTrace();

}

   final long fileLen = fs.getFileStatus(new Path(filename)).getLen();     

   String range=request.getHeader("Range");

   response.setHeader("Content-type","video/mp4");

   OutputStream out=response.getOutputStream();    

     if(range==null)

     {

    filename=fpath.substring(fpath.lastIndexOf("/")+1);

    response.setHeader("Content-Disposition", "attachment; filename="+filename);

    response.setContentType("application/octet-stream");

    response.setContentLength((int)fileLen);

    IOUtils.copyBytes(in, out, fileLen, false);

     

     }

     else

     {

    long start=Integer.valueOf(range.substring(range.indexOf("=")+1, range.indexOf("-")));

    long count=fileLen-start;

    long end;

    if(range.endsWith("-"))

      end=fileLen-1;

    else

      end=Integer.valueOf(range.substring(range.indexOf("-")+1));

    String ContentRange="bytes "+String.valueOf(start)+"-"+end+"/"+String.valueOf(fileLen);

    response.setStatus(206);

    response.setContentType("video/mpeg4");

    response.setHeader("Content-Range",ContentRange);

    in.seek(start);

    try{

        IOUtils.copyBytes(in, out, count, false);

      }

    catch(Exception e)

      {

      throw e;

      }

     }

        in.close();

        in = null;

        out.close();

        out = null;

}

/**

* 查询/input/upload路径下的所有MP4格式的视频文件

* @param request

* @param response

* @return

* @throws Exception

*/

@RequestMapping(value = "/getAllVideo.do", method = {RequestMethod.POST, RequestMethod.GET})

@ResponseBody

public ArrayList<String> getDirectoryFromHdfs(HttpServletRequest request, HttpServletResponse response) throws Exception {

ArrayList<String> list = new ArrayList<String>();

String direPath = "/input/upload";

//创建configuration对象conf

        Configuration conf = new Configuration();

        //指定hdfs的nameservice为cluster1,是NameNode的URI-Uniform Resource Locator

        conf.set("fs.defaultFS","hdfs://ns");

        //指定hdfs的nameservice为cluster1

        conf.set("dfs.nameservices","ns");

        //cluster1下面有两个NameNode，分别是nna，nns active/standby

        conf.set("dfs.ha.namenodes.ns","nna,nns");

        //nna的RPC通信地址

        conf.set("dfs.namenode.rpc-address.ns.nna","199.66.68.111:8020");

        //nns的RPC通信地址

        conf.set("dfs.namenode.rpc-address.ns.nns","199.66.68.112:8020");

        //配置失败自动切换实现方式

        conf.set("dfs.client.failover.proxy.provider.ns", "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider");

FileSystem fs = FileSystem.get(URI.create(direPath), conf);

FileStatus[] filelist = fs.listStatus(new Path(direPath));

for (int i = 0; i < filelist.length; i++) {

FileStatus fileStatus = filelist[i];

// /input/upload路径下的文件名

String nameStrings = fileStatus.getPath().getName().toString();

if(nameStrings.substring(nameStrings.length()-3).equals("mp4")) {

list.add(fileStatus.getPath().getName());

}

}

fs.close();

return list;

}

/**

* 从hdfs中下载视频到本地

* @param request

* @param response

* @throws Exception

*/

@RequestMapping(value = "/downloadVideo.do", method = {RequestMethod.POST, RequestMethod.GET})

@ResponseBody

public void test111(HttpServletRequest request, HttpServletResponse response) throws Exception{

Configuration conf = new Configuration();

        //指定hdfs的nameservice为cluster1,是NameNode的URI-Uniform Resource Locator

        conf.set("fs.defaultFS","hdfs://ns");

        //指定hdfs的nameservice为cluster1

        conf.set("dfs.nameservices","ns");

        //cluster1下面有两个NameNode，分别是nna，nns active/standby

        conf.set("dfs.ha.namenodes.ns","nna,nns");

        //nna的RPC通信地址

        conf.set("dfs.namenode.rpc-address.ns.nna","199.66.68.111:8020");

        //nns的RPC通信地址

        conf.set("dfs.namenode.rpc-address.ns.nns","199.66.68.112:8020");

        //配置失败自动切换实现方式

        conf.set("dfs.client.failover.proxy.provider.ns", "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider");

        

FileSystem fs = FileSystem.get(conf);

// 获取输入流

FSDataInputStream fis = fs.open(new Path("/input/upload/dispose.mp4"));

// 创建输出流

FileOutputStream fos = new FileOutputStream(new File("D://测试/dispose.mp4"));

IOUtils.copyBytes(fis, fos, conf);

}

private static Configuration hbaseConf = HBaseConfiguration.create();

static {

        hbaseConf.set("fs.defaultFS","hdfs://199.66.68.112:8020");

        hbaseConf.set("hbase.zookeeper.quorum","199.66.68.111,199.66.68.112,199.66.68.113,199.66.68.114,199.66.68.115,199.66.68.116,199.66.68.117");

        hbaseConf.set("hbase.zookeeper.property.clientPort", "2181");

    }

/**

* 图片显示

* rowkey：hbase中的行键

* tablename：hbase的表名

* @param request

* @param response

* @throws IOException

*/

@RequestMapping(value = "/playImg.do", method = {RequestMethod.POST, RequestMethod.GET})

@ResponseBody

public void scanTable(HttpServletRequest request, HttpServletResponse response) throws IOException {

//行键

String rowKey = request.getParameter("rowKey")==null?"":request.getParameter("rowKey");

//表名

String tablename = request.getParameter("tablename")==null?"":request.getParameter("tablename");

//列簇

String family = request.getParameter("family")==null?"":request.getParameter("family");

//列名

String col = request.getParameter("col")==null?"":request.getParameter("col");

Configuration configuration = HBaseConfiguration.create();

        //configuration.set("fs.defaultFS","hdfs://199.66.68.112:8020");

        configuration.set("hbase.zookeeper.quorum","199.66.68.111,199.66.68.112,199.66.68.113,199.66.68.114,199.66.68.115,199.66.68.116,199.66.68.117");

        configuration.set("hbase.zookeeper.property.clientPort", "2181");

        

        

        HTable table = new HTable(configuration,tablename);

        Get get = new Get(rowKey.getBytes());

        get.addColumn(Bytes.toBytes(family),Bytes.toBytes(col));

        Result rs = table.get(get);

        

        if(!rs.isEmpty()) {

        //bs为图片的字节流

        //保存get result的结果，字节数组形式

        byte[] bs = rs.getValue(Bytes.toBytes(family), Bytes.toBytes(col));

            table.close();

            

            OutputStream out=response.getOutputStream();

            out.write(bs);

            out.flush();

            out.close();

        }

   

        

    }

}