/**

 * 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）

Local

file

fs.LocalFileSystem

HDFS

hdfs

hdfs.DistrbutedFilesSystem

HFTP

hftp

hdfs.HftpFilesSystem

HSFTP

hsftp

hdfs.HsftpFilesSystem

HAR

har

fs.HarFileSystem

KFS

kfs

fs.kfs.KosmosFilesSystem

FTP

ftp

Fs.ftp.FtpFileSystem

//用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
[命令选项]     
帮助