ICode9
精准搜索请尝试： 精确搜索
首页
编程语言>
数据库
系统相关
互联网
其他分享
Python
Java
JavaScript
android
PHP
首页 > 其他分享> 文章详细

Hadoop 概述(三)

2022-05-04 14:36:15  阅读：178  来源： 互联网

标签：Hadoop hadoop job 概述 new apache org public


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

编译方法：

  1. 修改pom.xml中的hadoop版本信息和IntelliJ的安装目录
    macos的安装目录应该在/Applications/IntelliJ IDEA.app/Contents

  2. 先后执行如下命令

mvn clean
mvn assembly:assembly
  1. 在项目的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
  1. 用户自定义的mapper要继承自己的父类
  2. mapper的输入数据是KV对的形式
  3. mapper中的业务逻辑写在map()方法中
  4. mapper的输出数据是kv对的形式
  5. map()方法(MapTask进程)对每一个<K,V>调用一次
  • reducetask
  1. 用户自定义的Reducer要继承自己的父类
  2. Reducer的输入数据类型对应mapper的输出数据类型，也是KV
  3. Reducer的业务逻辑写在reduce()方法中
  4. 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类，我们可以到我们的输出路径上看到我们的结果，是按手机号码排序的一个结果
如下图：
输入文件：

输出文件：

总结

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

  1. 必须实现WritableComparable接口
  2. 反序列化时，需要反射空构造函数，必须有空构造函数
  3. 重写序列化write方法
  4. 重写反序列化readFields方法
  5. 注意反序列化的顺序和序列化顺序完全一致
  6. 要想把结果显示在文件中，需要重写toString(),可用'\t'分开，方便后续使用。
  7. 如果需要将自定义的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的文件信息，说明我们的分区的操作生效了。

总结：

  • 具体步骤如下：
  1. 自定义类继承Partitioner，重写getPartition()方法
  2. 在job驱动中，设置自定义Partitioner
  3. 自定义Partition后，要根据自定义Partitoner的逻辑设置相应数量的ReduceTask
  • partition分区总结
  1. 如果reduceTask的数量 > getPartition()的结果数，则会多产生几个空的输出文件part-r-000xx;
  2. 如果1 < ReduceTask的数量 < getPartition()的结果数，则有一部分分区数据无处安放，会报Exception;
  3. 如果ReduceTask的数量 = 1，则不管MapTask端输出多少个分区文件，最终结果都交给一个ReduceTask，最终也就只会有一个结果文件
  4. 分区号必须从0开始，逐一累加

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

 

 

标签：Hadoop,hadoop,job,概述,new,apache,org,public
来源： https://www.cnblogs.com/hanease/p/16221028.html

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

关于我们 | 联系我们 | 留言反馈

专注分享技术，共同学习，共同进步。侵权联系[81616952@qq.com]

Copyright (C)ICode9.com, All Rights Reserved.

ICode9版权所有