Hadoop 概述(三)

HDFS shell API

HDFS作为大数据的文件系统，可以放置数据文件，列举几个常用的shell脚本命令，
用法和linux中的基本类似，不过这个是hadoop里的一套，所以我们要用hadoop fs开头
具体可以使用hadoop fs --help 查看帮助信息

例如：
ls
hadoop fs -ls /

mkdir
hadoop fs -mkdir -p /hdfsapi/test

touch
hadoop fs -touch /hdfsapi/test/a.txt

chmod
hadoop fs -chmod -R 777 /hdfsapi/test/a.txt

chown
hadoop fs -chown -R HADOOP:HADOOP /hdfsapi

HDFS JAVA API

先下载一个hadoop hdfs插件,github上有源码，需要自己编译。
github地址：https://github.com/fangyuzhong2016/HadoopIntellijPlugin

编译方法：

修改pom.xml中的hadoop版本信息和IntelliJ的安装目录
macos的安装目录应该在/Applications/IntelliJ IDEA.app/Contents
先后执行如下命令

mvn clean
mvn assembly:assembly

在项目的target目录下就能找到一个zip的压缩包，这个就是插件，可以安装到idea中去

JAVA操作HDFS API

1.创建一个maven工厂

  ##pom.xml 文件配置信息如下
    <dependencies>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>3.1.4</version>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
            <scope>test</scope>
        </dependency>
    </dependencies>

2.创建一个test类

 public class ApiTest {

    public static final String HDFS_PATH = "hdfs://hadoop01:8020";
    //文件系统
    FileSystem fileSystem = null;
    //配置类
    Configuration configuration = null;

    //Before适用于类加载之前
    @Before
    public void setUp() throws Exception {
        configuration = new Configuration(true);
        //拿到文件系统
        fileSystem = FileSystem.get(new URI(HDFS_PATH), configuration, "hadoop");
    }

    //关闭资源用的这个
    @After
    public void tearDown() throws Exception {
        //释放资源
        configuration = null;
        fileSystem = null;
    }

    /**
     * 创建HDFS目录
     */
    @Test
    public void mkdir() throws Exception {
        fileSystem.mkdirs(new Path("/hdfsapi/test"));
    }

    /**
     * 创建文件
     */
    @Test
    public void create() throws Exception {
        FSDataOutputStream output = fileSystem.create(new Path("/hdfsapi/test/a.txt"));
        output.write("hello hadoop".getBytes());
        output.flush();
        output.close();
    }

    /**
     * 查看HDFS文件上的内容
     */
    @Test
    public void cat() throws Exception {
        FSDataInputStream in = fileSystem.open(new Path("/hdfsapi/test/a.txt"));
        IoUtils.copyBytes(in, System.out, 1024);
        in.close();
    }
    
    /**
     * 重命名文件
     */
    @Test
    public void rename() throws Exception {
        Path oldpath = new Path("/hdfsapi/test/a.txt");
        Path newPath = new Path("/hdfsapi/test/b.txt");
        fileSystem.rename(oldpath, newPath);
    }

    /**
     * 上传一个文件
     *
     * @throws Exception
     */
    @Test
    public void copyFromLocalFile() throws Exception {
        Path localPath = new Path("/Users/waterair/Documents/workspace/hadoop-3.1.4-src.tar.gz");
        Path hdfsPath = new Path("/hdfsapi/test");
        fileSystem.copyFromLocalFile(localPath, hdfsPath);
    }

    /**
     * 上传一个大文件
     *
     * @throws Exception
     */
    @Test
    public void copyFromLocalBigFile() throws Exception {
        InputStream in = new BufferedInputStream(
                new FileInputStream(
                        new File("/Users/waterair/Documents/workspace/hadoopIntelliJ/target/HadoopIntellijplugin-1.0.zip")));

        FSDataOutputStream output = fileSystem.create(new Path("/hdfsapi/test/all.zip"),
                new Progressable() {
                    public void progress() {
                        System.out.print(".");  //带进度提醒信息
                    }
                });


        IoUtils.copyBytes(in, output, 4096);
    }
    /**
     * 下载HDFS文件
     */
    @Test
    public void copyTOLocalFile() throws Exception{
        Path localPath = new Path("/Users/waterair/Documents/workspace/b.txt");
        Path hdfsPath = new Path("/hdfsapi/test/b.txt");
        fileSystem.copyToLocalFile(false,hdfsPath,localPath,true);
    }

    /**
     * 列举目录下所有文件信息
     * @throws Exception
     */
    @Test
    public void listFiles() throws Exception {
        FileStatus[] fileStatuses = fileSystem.listStatus(new Path("/hdfsapi/test"));

        for (FileStatus fileStatus : fileStatuses) {
            String isDir = fileStatus.isDirectory() ? "文件夹" : "文件";
            //副本
            short replication = fileStatus.getReplication();
            //大小
            long len = fileStatus.getLen();
            //路径
            String path = fileStatus.getPath().toString();

            System.out.println(isDir + "\t" + replication + "\t" + len + "\t" + path);
        }
    }

    /**
     * 展示HDFS中块信息
     * @throws Exception
     */
    @Test
    public void blks() throws Exception {
        Path i = new Path("/hdfsapi/test/test.txt");
        FileStatus fileStatus = fileSystem.getFileStatus(i);
        BlockLocation[] blks = fileSystem.getFileBlockLocations(fileStatus, 0, fileStatus.getLen());
        for (BlockLocation blk : blks) {
            System.out.println(blk);
        }
    }

    /**
     * 删除HDFS目录下所有
     * @throws Exception
     */
    @Test
    public void delete() throws Exception{
        fileSystem.delete(new Path("/hdfsapi"),true);
    }
}

--------------------------------------------------------------------------

hadoop wordCount小案例

接上回我们谈的hdfs api相关的操作，我们来先跑一个小案例，后边详细说mapreduce和yarn的原理。

打开idea新建一个maven工程，pom文件中添加项目依赖如下：

 <dependencies>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>3.1.4</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>1.8.0-beta4</version>
        </dependency>
    </dependencies>

在src/resources目录下创建一个log4j.properties文件，添加日志打印信息

log4j.rootLogger=INFO,stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n
log4j.appender.logfile=org.apache.log4j.FileAppender
log4j.appender.logfile.File=target/hadoop.log
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

MapTask类

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/**
 * @author tenic
 * @version 1.0
 * @Date 2021-04-10
 * @Desc
 */
public class MyMapper extends Mapper<LongWritable,Text, Text,IntWritable> {

    private IntWritable num = new IntWritable(1);
    private Text outK = new Text();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //value为一行一个单词场景
        // context.write(value,num);

        //value 为一行多个单词场景,且用空格分割场景
        String line = value.toString();
        String[] strs = line.split(" ");
        for (String str : strs) {
            outK.set(str);
            context.write(outK,num);
        }
    }
}

ReduceTask类

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

/**
 * @author tenic
 * @version 1.0
 * @Date 2021-04-10
 * @Desc
 */
public class MyReducer extends Reducer<Text, IntWritable, Text,IntWritable> {

    private IntWritable outV = new IntWritable();
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws Exception {
        int sum = 0;
       //values 是相同key的一组数据，eg:[1,1,1,1,1,1,1]
       //加和就可以确定有多少个相同的key，即该key出现多少次
        for (IntWritable value : values) {
            sum += 1;
        }
        outV.set(sum);
        context.write(key,outV);
    }
}

Driver类

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;


/**
 * @author tenic
 * @version 1.0
 * @Date 2021-04-10
 * @Desc
 */
public class MapReduceDriver {

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration config = new Configuration();
        //创建job任务
        Job job = Job.getInstance(config);
        
        //设置jar启动类
        job.setJarByClass(MapReduceDriver.class);
        
        //设置自己的mapper/reducer类
        job.setMapperClass(MyMapper.class);
        job.setReducerClass(MyReducer.class);
        
        //设置maptask最后输出的key/value的类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        
        //设置最后输出的key/value的类型
        job.setoutputKeyClass(Text.class);
        job.setoutputValueClass(IntWritable.class);
        
        //设置要进行wordcount的文件的路径
        FileInputFormat.setInputPaths(job,new Path("/Users/waterair/Documents/shelltest/mr/demo1/input/demo1.txt"));
        
        //设置最后wordcount结果的输出路径
        FileOutputFormat.setoutputPath(job,new Path("/Users/waterair/Documents/shelltest/mr/demo1/output1/"));
        
        //发布任务，并待任务执行完后，退出程序
        System.exit(job.waitForCompletion(true)?0:1);
    }
}

local模式

比如我们的想要进行wordcount 的文件内容如下：

直接运行这个Driver类，如果运行成功，就可以到我们的输出路径下找到我们的结果。
具体结果如下：

由于我们是在没有启动集群的基础上，直接运行idea中的程序，hadoop对这种情况称为local模式，也可以从打印的日志中发现如下信息：

2021-04-12 21:46:15,900 DEBUG [org.apache.hadoop.mapreduce.Cluster] - Cannot pick org.apache.hadoop.mapred.YarnClientProtocolProvider as the ClientProtocolProvider - returned null protocol
……
2021-04-12 21:46:23,073 DEBUG [org.apache.hadoop.mapreduce.Cluster] - Picked org.apache.hadoop.mapred.LocalClientProtocolProvider as the ClientProtocolProvider

集群模式

想要在我们部署好的环境中运行wordcount小案例，我们需要调整一下文件的输入路径和输出路径。
因为我们是在hadoop里正式运行，所以想要的文件路径应该对应到hadoop的hdfs中，我们使用args[]进行传递入参。
修改上边的Driver类中的代码，修改inputpath/outputpath：

        FileInputFormat.setInputPaths(job,new Path(args[0]));
        FileOutputFormat.setoutputPath(job,new Path(args[1]));

将项目打jar包出来，上传到我们的hadoop集群，要进行wordcount的文件也上传到我们的集群中

分别执行如下命令，将要进行wordcount的文件上传到hdfs，和启动hadoop的mapreduce任务

[hadoop@hadoop01 hadoop-3.1.4]$ ls
bin        etc          include  libexec      logs        README.txt  share
demo1.txt  hadoopDatas  lib      LICENSE.txt  NOTICE.txt  sbin        wc.jar
[hadoop@hadoop01 hadoop-3.1.4]$ hadoop fs -put ./demo1.txt /input
[hadoop@hadoop01 hadoop-3.1.4]$ hadoop fs -ls /
Found 3 items
-rw-r--r--   3 hadoop supergroup        247 2021-04-12 22:43 /input
drwxrwx---   - hadoop supergroup          0 2021-03-07 23:35 /tmp
drwxr-xr-x   - hadoop supergroup          0 2021-03-20 17:41 /user
[hadoop@hadoop01 hadoop-3.1.4]$ hadoop jar ./wc.jar com.tenic.hadoop.mr.demo2.MapReduceDriver /input /output
[hadoop@hadoop01 hadoop-3.1.4]$ hadoop fs -ls /
[hadoop@hadoop01 hadoop-3.1.4]$ hadoop fs -cat /output/part-r-00000

如果一切顺利，我们查看到了最后的输出文件和我们local模式的是一样的结果

总结：

代码的具体逻辑如下：

maptask

用户自定义的mapper要继承自己的父类
mapper的输入数据是KV对的形式
mapper中的业务逻辑写在map()方法中
mapper的输出数据是kv对的形式
map()方法(MapTask进程)对每一个<K,V>调用一次

reducetask

用户自定义的Reducer要继承自己的父类
Reducer的输入数据类型对应mapper的输出数据类型，也是KV
Reducer的业务逻辑写在reduce()方法中
ReduceTask进行对每一组相同K的<K,V>组调用一次reduce()方法

driver部分
相当于YARN集群的客户端，用于提交我们整个程序到YARN集群，提交的是封装了MapReduce程序相关运行参数的job对象

--------------------------------------------------------------------------

hadoop 序列化小案例

上次我们做了简单的一个wordcount案例，下边我们在做一个复杂一点的小案例。
例如传输一个对象到wordcount中，输入数据格式如下：

3	12939234290		120.156.21.100		www.baidu.com  	 116	           125		 	200
id       手机号码                       网络ip               网址         下载流量           上传流量                 响应状态码

期望输出的数据格式：

12939234290         116            125        241
手机号码             下行流量        上行流量    总流量

hadoop支持大量数据进行分布式存储，便于计算向数据移动。但是上边的数据是多个，但在hadoop中并没有支持对象这种复杂型数据类型，那该如何进行传输呢？
那我们对应上边的数据进行封装，那封装的对象该如何传递呢？
这里是不是会想到RPC框架里的数据传递，使用序列化。对的，hadoop也使用了自己的序列化方案。
那它和我们普通的序列化比较有何不同呢？
1.紧凑：存储空间更少
2.快速：传输速度更快
3.互操作：支持多语言操作

分析：
我们需要将上边一条数据，利用Tab分隔符进行分割，得到每一个数据，
然后对其进行封装到我们自定义的一个对象中去。
那自定义的对象要如何，序列化呢？hadoop有没有现成的序列化和反序列化呢？
我们可以查看之前小案例中的IntWriteable或者LongWriteable类，
他们可以在hadoop的序列化和反序列化中传递，都实现了一个相同的接口WritableComparable接口。
在进入查看WritableComparable接口，可以看到是继承了Writable和Comparable接口。这2个接口都有什么方法呢？
在Writable接口中，我们看到了序列化和反序列化的方法注释和解释
在Comparable接口中，我们看到了排序的方法注释和解释，针对我们的key进行排序
那我们就仿照这这个来定义一下我们要传递的对象

FlowBean类

import org.apache.hadoop.io.WritableComparable;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

/**
 * @author tenic
 * @version 1.0
 * @Date 2021-04-13
 * @Desc
 */
public class FlowBean  implements WritableComparable<FlowBean> {

    private Long downFlow;
    private Long upFlow;
    private Long totalFlow;

    public Long getUpFlow() {
        return upFlow;
    }

    public void setUpFlow(Long upFlow) {
        this.upFlow = upFlow;
    }

    public Long getDownFlow() {
        return downFlow;
    }

    public void setDownFlow(Long downFlow) {
        this.downFlow = downFlow;
    }

    public Long getTotalFlow() {
        return totalFlow;
    }

    public void setTotalFlow(Long totalFlow) {
        this.totalFlow = totalFlow;
    }

    public FlowBean() {
        super();
    }

    public void setSumFlow(){
        this.totalFlow=this.upFlow + this.downFlow;
    }

    @Override
    public void write(DataOutput out) throws IOException {
        out.writeLong(this.downFlow);
        out.writeLong(this.upFlow);
        out.writeLong(this.totalFlow);
    }

    @Override
    public void readFields(DataInput in) throws IOException {
        this.downFlow = in.readLong();
        this.upFlow = in.readLong();
        this.totalFlow = in.readLong();
    }

    @Override
    public int compareto(FlowBean o) {
        return (this.totalFlow <o.totalFlow ? -1 : (this.totalFlow ==o.totalFlow ? 0 : 1));
    }

    @Override
    public String toString() {
        return  downFlow + "\t" + upFlow + "\t" + totalFlow ;
    }
}

FlowMapper类

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/**
 * @author tenic
 * @version 1.0
 * @Date 2021-04-13
 * @Desc
 */
public class FlowMapper extends Mapper<LongWritable,Text, Text,FlowBean> {

    private FlowBean flowBean = new FlowBean();
    private Text outK = new Text();


    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();
        String[] args = line.split("\t");
        //1	12309866412		192.168.56.1	www.baidu.com		10	30	200
        //一行数据形式如上边所示，经过制表符分割后形成的数组形式里，
        // 下标为1的是手机号码，下标是4的是下载流量，下标是5的是上传流量
        outK.set(args[1]);
        flowBean.setDownFlow(Long.parseLong(args[args.length-3]));
        flowBean.setUpFlow(Long.parseLong(args[args.length-2]));
        flowBean.setSumFlow();

        context.write(outK,flowBean);
    }
}

FlowReducer类

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

/**
 * @author tenic
 * @version 1.0
 * @Date 2021-04-13
 * @Desc
 */
public class FlowReducer extends Reducer<Text,FlowBean, Text,FlowBean> {

    @Override
    protected void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException {
        // 18732919322   [{"upFlow":203,"downFlow":123,"totalFlow":326},{……}]
        //上边的是mapper输出的结果信息，我们可以看到前边是手机号，后边是一个flowbean的list对象
        Long upFlow = 0L;
        Long downFlow = 0L;
        for (FlowBean value : values) {
            upFlow += value.getUpFlow();
            downFlow += value.getDownFlow();
        }
        FlowBean flowBean = new FlowBean();
        flowBean.setDownFlow(downFlow);
        flowBean.setUpFlow(upFlow);
        flowBean.setSumFlow();

        context.write(key,flowBean);

    }
}

FlowBeanDriver类

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;


/**
 * @author tenic
 * @version 1.0
 * @Date 2021-04-13
 * @Desc
 */
public class FlowBeanDriver {

    public static void main(String[] args) throws Exception {
        Configuration config = new Configuration();
        Job job = Job.getInstance(config);
        //设置启动类
        job.setJarByClass(FlowBeanDriver.class);
        //设置mapper和reducer类
        job.setMapperClass(FlowMapper.class);
        job.setReducerClass(FlowReducer.class);
        //设置mapper输出的key和value类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(FlowBean.class);
        //设置最后reducer输出的key和value的类型
        job.setoutputKeyClass(Text.class);
        job.setoutputValueClass(FlowBean.class);
        //输入输出路径信息
        FileInputFormat.setInputPaths(job,new Path("/Users/waterair/Documents/shelltest/mr/demo3/input/demo1.txt"));
        FileOutputFormat.setoutputPath(job,new Path("/Users/waterair/Documents/shelltest/mr/demo3/output1/"));

        System.exit(job.waitForCompletion(true)?0:1);
    }
}

运行我们的driver类，我们可以到我们的输出路径上看到我们的结果，是按手机号码排序的一个结果
如下图：
输入文件：

输出文件：

总结

具体实现对象序列化步骤如下：

必须实现WritableComparable接口
反序列化时，需要反射空构造函数，必须有空构造函数
重写序列化write方法
重写反序列化readFields方法
注意反序列化的顺序和序列化顺序完全一致
要想把结果显示在文件中，需要重写toString(),可用'\t'分开，方便后续使用。
如果需要将自定义的bean放在key中传输，则还需要实现Comparable接口的方法，
因为MapReduce框架中有排序的过程，必须要按某一个key进行升序排列

--------------------------------------------------------------------------

hadoop 自定义分区

自定义分区

我们在wordcount小案例中看到结果是1个part-r-000000的文件，那如果我想对统计结果，
按照不同的条件输出到不同的文件(分区)，那该如何处理呢？

我们梳理一下这个过程先
一个文本文件，上传到hdfs后以block块存储，split到切片，一个切片对应一个maptask任务，
一个maptask任务会对数据进行分区、归并和排序等操作，输出成一个临时文件（外部无序，内部有序），
一个分区对应一个reducetask，按相同的key为一组进行处理，最后输出到一个文件中。

我们是不是就可以在maptask的时候就进行分区操作，后边reducetask就会处理相同分区的数据
那hadoop是怎么实现这个呢？我们在代码里可以看到有个Partitioner的抽象类，可以看到代码逻辑：

/**
 * 默认使用的是hashpartitoner分区，
 * 逻辑就是按照key的hashcode对numReduceTasks取模得到分区号
**/
public class HashPartitioner<K, V> extends Partitioner<K, V> {

  /** Use {@link Object#hashCode()} to partition. */
  public int getPartition(K key, V value,
                          int numReduceTasks) {
    return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
  }
}

那我们是不是可以继承Partitioner来实现自己的业务逻辑呢？

具体步骤如下：

自定义类继承Partitioner，重写getPartition()方法

public class MyPartitioner extends Partitioner<Text,FlowBean> {
    @Override
    public int getPartition(Text text, FlowBean flowBean, int numPartitions) {
        //要返回的分区号
        int partition = 0;
        // maptask输出的key，也就是上个序列化例子里的手机号码
        String phonestr = text.toString();
        //截取手机号码前3位，来区分不同的分区
        String phoneNum = phonestr.substring(0, 3);
        if("131".equals(phoneNum)){
            partition =  0;
        }else if("132".equals(phoneNum)){
            partition =  1;
        }else if("133".equals(phoneNum)){
            partition =  2;
        }else if("134".equals(phoneNum)){
            partition =  3;
        }else {
            partition = 4;
        }
        return partition;
    }
}

FlowBeanDriver 驱动类

public class FlowBeanDriver {

    public static void main(String[] args) throws Exception {
        Configuration config = new Configuration();
        Job job = Job.getInstance(config);
        //设置启动类
        job.setJarByClass(FlowBeanDriver.class);
        //设置mapper和reducer类
        job.setMapperClass(FlowMapper.class);
        job.setReducerClass(FlowReducer.class);
        //设置mapper输出的key和value类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(FlowBean.class);

        ==//2.在job驱动中，设置自定义Partitioner==
        job.setPartitionerClass(MyPartitioner.class);
        ==//3.自定义Partition后，要根据自定义Partitoner的逻辑设置相应数量的ReduceTask==
        job.setNumReduceTasks(5);

        //设置最后reducer输出的key和value的类型
        job.setoutputKeyClass(Text.class);
        job.setoutputValueClass(FlowBean.class);
        //输入输出路径信息
        FileInputFormat.setInputPaths(job,new Path("/Users/waterair/Documents/shelltest/mr/demo4/input/demo1.txt"));
        FileOutputFormat.setoutputPath(job,new Path("/Users/waterair/Documents/shelltest/mr/demo4/output1/"));

        System.exit(job.waitForCompletion(true)?0:1);
    }
}

然后运行，发现我们的输出目录下会有多个part-r-00000x的文件信息，说明我们的分区的操作生效了。

总结：

具体步骤如下：

自定义类继承Partitioner，重写getPartition()方法
在job驱动中，设置自定义Partitioner
自定义Partition后，要根据自定义Partitoner的逻辑设置相应数量的ReduceTask

partition分区总结

如果reduceTask的数量 > getPartition()的结果数，则会多产生几个空的输出文件part-r-000xx;
如果1 < ReduceTask的数量 < getPartition()的结果数，则有一部分分区数据无处安放，会报Exception;
如果ReduceTask的数量 = 1，则不管MapTask端输出多少个分区文件，最终结果都交给一个ReduceTask，最终也就只会有一个结果文件
分区号必须从0开始，逐一累加

--------------------------------------------------------------------------

HDFS shell API

HDFS JAVA API

JAVA操作HDFS API

hadoop wordCount小案例

MapTask类

ReduceTask类

Driver类

local模式

集群模式

总结：

hadoop 序列化小案例

FlowBean类

FlowMapper类

FlowReducer类

FlowBeanDriver类

总结

hadoop 自定义分区

自定义分区

总结：

相关推荐