/**
 * Created by 鸣宇淳 on 2016/3/23.
 */
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import java.io.IOException;
import java.util.StringTokenizer;

/*
功能：OrderDemo用来统计订单中下到各个经销商的订单数量，无主订单不统计
文件：订单信息在orderdata.txt文件中，文件中信息存储各个是：
格式：手机号      经销商ID   省份ID   城市ID   品牌ID   车型ID
例如：13146198476 3012   440000   440100   25     3467
分隔符：\t
 */
public class OrderDemo {

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

        //新建Hadoop配置类
        Configuration conf = new Configuration();

        //新建一个MapReduce Job，Job名字为OrderDemoJob，这个名字会显示在监控中
        Job job = new Job(conf, "OrderDemoJob");

        //设置MapReducer任务所用的jar，根据类找Jar包
        job.setJarByClass(OrderDemo.class);

        //设置Map方法所在类
        job.setMapperClass(Map.class);

//设置Combiner类

job.setCombinerClass(Combiner.class);

        //设置Reduce方法所在类
        job.setReducerClass(Reduce.class);

        //设置MapReduce Job输入数据格式
        job.setInputFormatClass(TextInputFormat.class);

        //设置输出数据格式
        job.setOutputFormatClass(TextOutputFormat.class);

        //设置MapReduce Job最终输出键的格式
        job.setOutputKeyClass(Text.class);

        //设置MapReduce Job最终输出值的格式
        job.setOutputValueClass(IntWritable.class);

        //设置输入文件路径
        FileInputFormat.addInputPath(job, new Path(args[0]));

        //设置输出文件路径
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        //提交任务给JobTracker，等待返回值
        job.waitForCompletion(true);
    }
    public static class Map extends Mapper<LongWritable, Text, Text, IntWritable> {
        public void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {
            String line = value.toString();
            String dealerid= line.split("\t")[1];
            if(!dealerid.equals("0"))//不计算无主订单
            {
                context.write(new Text(dealerid), new IntWritable(1));
            }
        }
    }
    public static class Reduce extends Reducer<Text, IntWritable, Text, IntWritable> {
        public void reduce(Text key, Iterable<IntWritable> values, Context context)
                throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable val : values) {
                sum += val.get();
            }
            context.write(key, new IntWritable(sum));
        }
    }

   public static class Combiner extends Reducer<Text, IntWritable, Text, IntWritable> {
    public void reduce(Text key, Iterable<IntWritable> values, Context context)
            throws IOException, InterruptedException {
        int sum = 0;
        for (IntWritable val : values) {
            sum += val.get();
        }
        context.write(key, new IntWritable(sum));
    }
}

}

[centos@hadoop1 ~]$ hadoop fs -mkdir /chybindemo/input

[centos@hadoop1~]$ hadoop fs -put /chybindemo/orderdata.txt /chybindemo/input

[centos@hadoop1~]$ hadoop  jar  /chybindemo/input/hadoopdemo.jar OrderDemo /chybindemo/input  /chybindemo/output

格式：

hadoop jar  [jar包路径]  [类名]  [输入路径]  [输出路径]

[centos@hadoop1 ~]$ hadoop fs -ls -R /chybindemo/output

[centos@hadoop1 ~]$ hadoop fs -cat /chybindemo/output/part-r-00000

文件系统

URI方案

Java实现（org.apache.hadoop）

//用URL接口读取HDFS中文件
static  {
    URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory() );
}
public  String  GetHDFSByURL(String url) throws MalformedURLException,IOException
{
    String str="";
    InputStream in =null;
    OutputStream out=null;
    try {
        in=new URL(url).openStream();
        //IOUtils.copyBytes(in,out,4096,false);
        str=out.toString();
    }
    finally {
        IOUtils.closeStream(in);
        IOUtils.closeStream(out);
    }
    return  str;
}

//ReadFile
//url:"/user/hadoop/data/write.txt"
public  String  ReadFile(String url)throws IOException
{
    String fileContent="";
    Configuration conf = new Configuration();
    FileSystem fs = FileSystem.get(conf);
    Path path = new Path(url);

    if(fs.exists(path)){
        FSDataInputStream is = fs.open(path);
        FileStatus status = fs.getFileStatus(path);
        byte[] buffer = new byte[Integer.parseInt(String.valueOf(status.getLen()))];
        is.readFully(0, buffer);
        is.close();
        fs.close();
        fileContent=buffer.toString();
    }
    return fileContent;
}

//创建HDFS目录
//dirpath： "/user/hadoop/data/20130709"
public  void  MakeDir(String dirpath) throws IOException {
    Configuration conf = new Configuration();
    FileSystem fs = FileSystem.get(conf);
    Path path = new Path(dirpath);
    fs.create(path);
    fs.close();
}

//HDFS写文件
//fileurl:"/user/hadoop/data/write.txt"
public  void  WriteFile(String fileurl,String fileContent)throws IOException
{
    Configuration conf = new Configuration();
    FileSystem fs = FileSystem.get(conf);
    Path path = new Path(fileurl);
    FSDataOutputStream out = fs.create(path);
    out.writeUTF(fileContent);
    fs.close();
}

//删除文件
//fileurl :"/user/hadoop/data/word.txt"
public void  DeleteFile(String fileurl)throws IOException
{
    Configuration conf = new Configuration();
    FileSystem fs = FileSystem.get(conf);

    Path path = new Path(fileurl);
    fs.delete(path,true);
    fs.close();
}

//查询文件的元数据
public  void  ShowFileStatus(String fileUrl) throws  IOException
{
    Configuration conf = new Configuration();
    FileSystem fs = FileSystem.get(conf);
    Path file=new Path(fileUrl);
    FileStatus stat=fs.getFileStatus(file);
    
    System.out.println("文件路径："+stat.getPath());
    System.out.println("是否是目录："+stat.isDirectory());
    System.out.println("是否是文件："+stat.isFile());
    System.out.println("块的大小："+stat.getBlockSize());
    System.out.println("文件所有者："+stat.getOwner()+":"+stat.getGroup());
    System.out.println("文件权限："+stat.getPermission());
    System.out.println("文件长度："+stat.getLen());
    System.out.println("备份数："+stat.getReplication());
    System.out.println("修改时间："+stat.getModificationTime());
}

-ls      -ls <路径>      查看指定路径的当前目录结构

-lsr      -lsr <路径>      递归查看指定路径的目录结构

-du      -du <路径>      统计目录下个文件大小

-dus      -dus <路径>      汇总统计目录下文件(夹)大小

-count      -count [-q] <路径>      统计文件(夹)数量

-mv      -mv <源路径> <目的路径>      移动

-cp      -cp <源路径> <目的路径>      复制

-rm      -rm [-skipTrash] <路径>      删除文件/空白文件夹

-rmr      -rmr [-skipTrash] <路径>      递归删除

-put      -put <多个linux上的文件> <hdfs路径>      上传文件

-copyFromLocal      -copyFromLocal <多个linux上的文件> <hdfs路径>      从本地复制

-moveFromLocal      -moveFromLocal <多个linux上的文件> <hdfs路径>      从本地移动

-getmerge      -getmerge <源路径> <linux路径>      合并到本地

-cat      -cat <hdfs路径>      查看文件内容

-text      -text <hdfs路径>      查看文件内容

-copyToLocal      -copyToLocal [-ignoreCrc] [-crc] [hdfs源路径] [linux目的路径]      从本地复制

-moveToLocal      -moveToLocal [-crc] <hdfs源路径> <linux目的路径>      从本地移动

-mkdir      -mkdir <hdfs路径>      创建空白文件夹

-setrep      -setrep [-R] [-w] <副本数> <路径>      修改副本数量

-touchz      -touchz <文件路径>      创建空白文件

-stat      -stat [format] <路径>      显示文件统计信息

-tail      -tail [-f] <文件>      查看文件尾部信息

-chmod      -chmod [-R] <权限模式> [路径]      修改权限

-chown      -chown [-R] [属主][:[属组]] 路径      修改属主

-chgrp      -chgrp [-R] 属组名称 路径      修改属组

-help      -help [命令选项]      帮助