package com.hadoop.base;


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

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.PathFilter;
/**
 * @function 将指定格式的多个文件上传至 HDFS
 *
 */
@SuppressWarnings("unused")
public class CopyManyFilesToHDFS {
    
    private static FileSystem fs = null;
    private static FileSystem local = null;

    /**
     * @function Main 方法
     * @param args
     * @throws IOException
     * @throws URISyntaxException
     */
    public static void main(String[] args) throws IOException,URISyntaxException {
        //文件上传路径
        Path dstPath = new Path("hdfs://master:9000/middle/filter/");
        //调用文件上传 list 方法
        list(dstPath);
    }

    /**
     * function 过滤文件格式   将多个文件上传至 HDFS
     * @param dstPath 目的路径
     * @throws IOException
     * @throws URISyntaxException
     */
    public static void list(Path dstPath) throws IOException, URISyntaxException {
        //读取hadoop文件系统的配置
        Configuration conf = new Configuration();
        //HDFS 接口
        URI uri = new URI("hdfs://master:9000");
        //获取文件系统对象
        fs = FileSystem.get(uri, conf);
        // 获得本地文件系统
        local = FileSystem.getLocal(conf);
        //只上传data/testdata 目录下 txt 格式的文件
        FileStatus[] localStatus = local.globStatus(new Path("D:/大数据/205/205_data/data/*"),new RegexAcceptPathFilter("^.*txt$"));
        // 获得所有文件路径
        Path[] listedPaths = FileUtil.stat2Paths(localStatus);
        for(Path p:listedPaths){
            //将本地文件上传到HDFS
            fs.copyFromLocalFile(p, dstPath);
        }
    }

    /**
     * @function 只接受 txt 格式的文件aa
     * @author
     *
     */
    public static class RegexAcceptPathFilter implements PathFilter {
        private final String regex;

        public RegexAcceptPathFilter(String regex) {
            this.regex = regex;
        }

        @Override
        public boolean accept(Path path) {
            // TODO Auto-generated method stub
            boolean flag = path.toString().matches(regex);
            //只接受 regex 格式的文件
            return flag;
        }
    }
}