AI 助理
备案控制台
开发者社区 大数据 文章 正文

Hadoop MapReduce编程 API入门系列之统计学生成绩版本2（十八）

2017-11-21 926
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
本文涉及的产品
全局流量管理 GTM，标准版 1个月
云解析 DNS，旗舰版 1个月
公共DNS（含HTTPDNS解析），每月1000万次HTTP解析
简介：

　　这里，为了空格符的差错，直接，我们有时候，像如下这样的来排数据。

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

代码

复制代码 
package zhouls.bigdata.myMapReduce.Gender;

import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
/**
* 
* @function 统计不同年龄段内 男、女最高分数
*
*
*/

/*
Alice<tab>23<tab>female<tab>45
Bob<tab>34<tab>male<tab>89
Chris<tab>67<tab>male<tab>97
Kristine<tab>38<tab>female<tab>53
Connor<tab>25<tab>male<tab>27
Daniel<tab>78<tab>male<tab>95
James<tab>34<tab>male<tab>79
Alex<tab>52<tab>male<tab>69
Nancy<tab>7<tab>female<tab>98
Adam<tab>9<tab>male<tab>37
Jacob<tab>7<tab>male<tab>23
Mary<tab>6<tab>female<tab>93
Clara<tab>87<tab>female<tab>72
Monica<tab>56<tab>female<tab>92
*/
public class Gender extends Configured implements Tool {
/*
* 
* @function Mapper 解析输入数据，然后按需求输出
* @input key=行偏移量 value=学生数据
* @output key=gender value=name+age+score
* 
*/
public static class PCMapper extends Mapper<Object, Text, Text, Text>
{
public void map(Object key, Text value, Context context) throws IOException, InterruptedException 
{//拿Alice<tab>23<tab>female<tab>45
String[] tokens = value.toString().split("<tab>");//使用分隔符<tab>，将数据解析为数组 tokens
//得到Alice    23    female    45
//即tokens[0] tokens[1] tokens[2] tokens[3] 
String gender = tokens[2].toString();//性别
String nameAgeScore = tokens[0] + "\t" + tokens[1] + "\t"+ tokens[3];
//输出 key=gender value=name+age+score
//输出 key=female value=Alice    +23+45
context.write(new Text(gender), new Text(nameAgeScore));//将 （female ， Alice+ 23+ 45) 写入到context中
}
}
public static class MyHashPartitioner extends Partitioner<Text, Text> 
{
/** Use {@link Object#hashCode()} to partition. */
@Override
public int getPartition(Text key, Text value,int numReduceTasks) 
{
return (key.hashCode()) % numReduceTasks;
}

}
/**
* 
* @function Partitioner 根据 age 选择 reduce 分区
*
*/
public static class PCPartitioner extends Partitioner<Text, Text> 
{

@Override
public int getPartition(Text key, Text value, int numReduceTasks) 
{
// TODO Auto-generated method stub
String[] nameAgeScore = value.toString().split("\t");
String age = nameAgeScore[1];//学生年龄
int ageInt = Integer.parseInt(age);//按年龄段分区

// 默认指定分区 0
if (numReduceTasks == 0)
return 0;

//年龄小于等于20，指定分区0
if (ageInt <= 20) {
return 0;
}
// 年龄大于20，小于等于50，指定分区1
if (ageInt > 20 && ageInt <= 50) {

return 1 % numReduceTasks;
}
// 剩余年龄，指定分区2
else
return 2 % numReduceTasks;
}
}

/**
* 
* @function 定义Combiner 合并 Mapper 输出结果
*
*/
public static class PCCombiner extends Reducer<Text, Text, Text, Text> 
{
private Text text = new Text();

public void reduce(Text key, Iterable<Text> values, Context context)throws IOException, InterruptedException 
{
int maxScore = Integer.MIN_VALUE;
String name = " ";
String age = " ";
int score = 0;
for (Text val : values) 
{
String[] valTokens = val.toString().split("\\t");
score = Integer.parseInt(valTokens[2]);
if (score > maxScore) 
{
name = valTokens[0];
age = valTokens[1];
maxScore = score;
}
}
text.set(name + "\t" + age + "\t" + maxScore);
context.write(key, text);
}
}

/*
* 
* @function Reducer 统计出 不同年龄段、不同性别 的最高分
* input key=gender value=name+age+score
* output key=name value=age+gender+score
* 
*/
static class PCReducer extends Reducer<Text, Text, Text, Text> 
{
@Override
public void reduce(Text key, Iterable<Text> values, Context context)throws IOException, InterruptedException 
{
int maxScore = Integer.MIN_VALUE;
String name = " ";
String age = " ";
String gender = " ";
int score = 0;
// 根据key，迭代 values 集合，求出最高分
for (Text val : values)
{
String[] valTokens = val.toString().split("\\t");
score = Integer.parseInt(valTokens[2]);
if (score > maxScore) 
{
name = valTokens[0];
age = valTokens[1];
gender = key.toString();
maxScore = score;
}
}
context.write(new Text(name), new Text("age- " + age + "\t" + gender + "\tscore-" + maxScore));
}
}

/**
* @function 任务驱动方法
* @param args
* @return
* @throws Exception
*/
@Override
public int run(String[] args) throws Exception 
{
// TODO Auto-generated method stub
Configuration conf = new Configuration();//读取配置文件

Path mypath = new Path(args[1]);
FileSystem hdfs = mypath.getFileSystem(conf);
if (hdfs.isDirectory(mypath)) 
{
hdfs.delete(mypath, true);
}

@SuppressWarnings("deprecation")
Job job = new Job(conf, "gender");//新建一个任务
job.setJarByClass(Gender.class);//主类
job.setMapperClass(PCMapper.class);//Mapper
job.setReducerClass(PCReducer.class);//Reducer

job.setPartitionerClass(MyHashPartitioner.class);
//job.setPartitionerClass(PCPartitioner.class);//设置Partitioner类
job.setNumReduceTasks(3);// reduce个数设置为3

job.setMapOutputKeyClass(Text.class);//map 输出key类型
job.setMapOutputValueClass(Text.class);//map 输出value类型

job.setCombinerClass(PCCombiner.class);//设置Combiner类

job.setOutputKeyClass(Text.class);//输出结果 key类型
job.setOutputValueClass(Text.class);//输出结果 value 类型

FileInputFormat.addInputPath(job, new Path(args[0]));// 输入路径
FileOutputFormat.setOutputPath(job, new Path(args[1]));// 输出路径
job.waitForCompletion(true);//提交任务
return 0;
}
/**
* @function main 方法
* @param args
* @throws Exception
*/
public static void main(String[] args) throws Exception
{
//    String[] args0 = {
//    "hdfs://HadoopMaster:9000/gender/gender.txt",
//    "hdfs://HadoopMaster:9000/out/partition/" };

String[] args0 = {
"./data/gender/gender.txt",
"./out/gender" };


int ec = ToolRunner.run(new Configuration(),new Gender(), args0);
System.exit(ec);
}
}
复制代码

 

 

 

 

　　　　或者

　　　　代码

复制代码 
package com.dajiangtai.hadoop.junior;

import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
/**
 * 
 * @function 统计不同年龄段内    男、女最高分数
 * @author zhouls
 *
 */
 
 /*
Alice<tab>23<tab>female<tab>45
Bob<tab>34<tab>male<tab>89
Chris<tab>67<tab>male<tab>97
Kristine<tab>38<tab>female<tab>53
Connor<tab>25<tab>male<tab>27
Daniel<tab>78<tab>male<tab>95
James<tab>34<tab>male<tab>79
Alex<tab>52<tab>male<tab>69
Nancy<tab>7<tab>female<tab>98
Adam<tab>9<tab>male<tab>37
Jacob<tab>7<tab>male<tab>23
Mary<tab>6<tab>female<tab>93
Clara<tab>87<tab>female<tab>72
Monica<tab>56<tab>female<tab>92
*/
public class Gender extends Configured implements Tool {
    /*
     * 
     * @function Mapper 解析输入数据，然后按需求输出
     * @input  key=行偏移量   value=学生数据
     * @output key=gender  value=name+age+score
     * 
     */
    public static class PCMapper extends Mapper<Object, Text, Text, Text>
    {
        public void map(Object key, Text value, Context context) throws IOException, InterruptedException 
        {//拿Alice<tab>23<tab>female<tab>45
            String[] tokens = value.toString().split("<tab>");//使用分隔符<tab>，将数据解析为数组 tokens
                            //得到Alice        23         female            45
                            //即tokens[0]   tokens[1]  tokens[2]  tokens[3] 
            String gender = tokens[2].toString();//性别
            String nameAgeScore = tokens[0] + "\t" + tokens[1] + "\t"+ tokens[3];
            //输出  key=gender  value=name+age+score
            //输出     key=female  value=Alice    +23+45
            context.write(new Text(gender), new Text(nameAgeScore));//将 （female ， Alice+  23+ 45) 写入到context中
        }
    }
    public static class MyHashPartitioner extends Partitioner<Text, Text> 
    {
          /** Use {@link Object#hashCode()} to partition. */
          @Override
          public int getPartition(Text key, Text value,int numReduceTasks) 
          {
            return (key.hashCode()) % numReduceTasks;
          }

        }
    /**
     * 
     * @function Partitioner 根据 age 选择 reduce 分区
     *
     */
    public static class PCPartitioner extends Partitioner<Text, Text> 
    {

        @Override
        public int getPartition(Text key, Text value, int numReduceTasks) 
        {
            // TODO Auto-generated method stub
            String[] nameAgeScore = value.toString().split("\t");
            String age = nameAgeScore[1];//学生年龄
            int ageInt = Integer.parseInt(age);//按年龄段分区

            // 默认指定分区 0
            if (numReduceTasks == 0)
                return 0;

            //年龄小于等于20，指定分区0
            if (ageInt <= 20) {
                return 0;
            }
            // 年龄大于20，小于等于50，指定分区1
            if (ageInt > 20 && ageInt <= 50) {

                return 1 % numReduceTasks;
            }
            // 剩余年龄，指定分区2
            else
                return 2 % numReduceTasks;
        }
    }

    /**
     * 
     * @function 定义Combiner 合并 Mapper 输出结果
     *
     */
    public static class PCCombiner extends Reducer<Text, Text, Text, Text> 
    {
        private Text text = new Text();

        public void reduce(Text key, Iterable<Text> values, Context context)throws IOException, InterruptedException 
        {
            int maxScore = Integer.MIN_VALUE;
            String name = " ";
            String age = " ";
            int score = 0;
            for (Text val : values) 
            {
                String[] valTokens = val.toString().split("\\t");
                score = Integer.parseInt(valTokens[2]);
                if (score > maxScore) 
                {
                    name = valTokens[0];
                    age = valTokens[1];
                    maxScore = score;
                }
            }
            text.set(name + "\t" + age + "\t" + maxScore);
            context.write(key, text);
        }
    }

    /*
     * 
     * @function Reducer 统计出 不同年龄段、不同性别 的最高分
     * input key=gender value=name+age+score
     * output key=name value=age+gender+score
     * 
     */
    static class PCReducer extends Reducer<Text, Text, Text, Text>  
    {
        @Override
        public void reduce(Text key, Iterable<Text> values, Context context)throws IOException, InterruptedException 
        {
            int maxScore = Integer.MIN_VALUE;
            String name = " ";
            String age = " ";
            String gender = " ";
            int score = 0;
            // 根据key，迭代 values 集合，求出最高分
            for (Text val : values)
                {
                String[] valTokens = val.toString().split("\\t");
                score = Integer.parseInt(valTokens[2]);
                if (score > maxScore) 
                {
                    name = valTokens[0];
                    age = valTokens[1];
                    gender = key.toString();
                    maxScore = score;
                }
            }
            context.write(new Text(name), new Text("age- " + age + "\t" + gender + "\tscore-" + maxScore));
        }
    }

    /**
     * @function 任务驱动方法
     * @param args
     * @return
     * @throws Exception
     */
    @Override
    public int run(String[] args) throws Exception 
    {
        // TODO Auto-generated method stub
        Configuration conf = new Configuration();//读取配置文件

        Path mypath = new Path(args[1]);
        FileSystem hdfs = mypath.getFileSystem(conf);
        if (hdfs.isDirectory(mypath)) 
        {
            hdfs.delete(mypath, true);
        }

        @SuppressWarnings("deprecation")
        Job job = new Job(conf, "gender");//新建一个任务
        job.setJarByClass(Gender.class);//主类
        job.setMapperClass(PCMapper.class);//Mapper
        job.setReducerClass(PCReducer.class);//Reducer

        job.setPartitionerClass(MyHashPartitioner.class);
        //job.setPartitionerClass(PCPartitioner.class);//设置Partitioner类
        job.setNumReduceTasks(3);// reduce个数设置为3

        job.setMapOutputKeyClass(Text.class);//map 输出key类型
        job.setMapOutputValueClass(Text.class);//map 输出value类型

        job.setCombinerClass(PCCombiner.class);//设置Combiner类

        job.setOutputKeyClass(Text.class);//输出结果 key类型
        job.setOutputValueClass(Text.class);//输出结果 value 类型

        FileInputFormat.addInputPath(job, new Path(args[0]));// 输入路径
        FileOutputFormat.setOutputPath(job, new Path(args[1]));// 输出路径
        job.waitForCompletion(true);//提交任务
        return 0;
    }
    /**
     * @function main 方法
     * @param args
     * @throws Exception
     */
    public static void main(String[] args) throws Exception
    {
        String[] args0 = {
                "hdfs://master:9000/middle/partition/gender.txt",
                "hdfs://master:9000/middle/partition/out/" };
        int ec = ToolRunner.run(new Configuration(),new Gender(), args0);
        System.exit(ec);
    }
}
复制代码

 



本文转自大数据躺过的坑博客园博客，原文链接：http://www.cnblogs.com/zlslch/p/6165704.html，如需转载请自行联系原作者

文章标签：
云解析DNS
API
Apache
分布式计算
Hadoop
关键词：
API入门
hadoop mapreduce
mapreduce hadoop
API编程
编程API
技术小哥哥
目录
相关文章
土木林森
|
3月前
|
安全 Java API
告别繁琐编码，拥抱Java 8新特性：Stream API与Optional类助你高效编程，成就卓越开发者！
【8月更文挑战第29天】Java 8为开发者引入了多项新特性，其中Stream API和Optional类尤其值得关注。Stream API对集合操作进行了高级抽象，支持声明式的数据处理，避免了显式循环代码的编写；而Optional类则作为非空值的容器，有效减少了空指针异常的风险。通过几个实战示例，我们展示了如何利用Stream API进行过滤与转换操作，以及如何借助Optional类安全地处理可能为null的数据，从而使代码更加简洁和健壮。
土木林森
106 0
bruce刘晓伟-18435
|
1月前
|
IDE API 定位技术
Python--API编程：IP地址翻译成实际的物理地址
Python--API编程：IP地址翻译成实际的物理地址
bruce刘晓伟-18435
53 0
游客qeoynko2nmbgk
|
2月前
|
网络协议 API Windows
MASM32编程调用 API函数RtlIpv6AddressToString，windows 10 容易，Windows 7 折腾
MASM32编程调用 API函数RtlIpv6AddressToString，windows 10 容易，Windows 7 折腾
游客qeoynko2nmbgk
28 0
土木林森
|
3月前
|
JavaScript API 开发者
RESTful API 设计的传奇征程：突破常规，拥抱最佳实践，铸就编程巅峰！
【8月更文挑战第7天】希望通过以上的探讨，能让您对 RESTful API 设计有更深入的理解和认识。
土木林森
52 5
土木林森
|
3月前
|
JSON API 数据库
神秘编程力量来袭！Rails 究竟隐藏着怎样的魔力，能构建出强大的 RESTful API？快来一探究竟！
【8月更文挑战第31天】《构建 RESTful API：使用 Rails 进行服务端开发》介绍了如何利用 Ruby on Rails 框架高效构建可扩展的 RESTful API。Rails 采用“约定优于配置”，简化开发流程，通过示例展示了路由定义、控制器设计及模型层交互等内容，帮助开发者快速搭建稳定可靠的服务端。无论小型项目还是大型应用，Rails 均能提供强大支持，提升开发效率。
土木林森
28 0
JavaPub
|
5月前
|
分布式计算 Hadoop Android开发
hadoop-eclipse-plugin(版本hadoop2.7.3)
hadoop-eclipse-plugin(版本hadoop2.7.3)
JavaPub
155 6
hadoop-eclipse-plugin(版本hadoop2.7.3)
听风de歌
|
5月前
|
存储 分布式计算 Hadoop
Hadoop 3.x版本的新特性
【6月更文挑战第18天】
听风de歌
150 2
JavaPub
|
5月前
|
分布式计算 Hadoop Java
运行Hadoop自带的wordcount单词统计程序
运行Hadoop自带的wordcount单词统计程序
JavaPub
113 3
JavaPub
|
5月前
|
分布式计算 Java Hadoop
简单的java Hadoop MapReduce程序(计算平均成绩)从打包到提交及运行
简单的java Hadoop MapReduce程序(计算平均成绩)从打包到提交及运行
JavaPub
55 0
武子康
|
1月前
|
分布式计算 Kubernetes Hadoop
大数据-82 Spark 集群模式启动、集群架构、集群管理器 Spark的HelloWorld + Hadoop + HDFS
大数据-82 Spark 集群模式启动、集群架构、集群管理器 Spark的HelloWorld + Hadoop + HDFS
武子康
132 6

热门文章

最新文章

  • 1
    Idea里面老版本MapReduce设置FileInputFormat参数格式变化
  • 2
    MapReduce的模式，算法以及用例
  • 3
    MapReduce实现WordCount
  • 4
    Hadoop2.6.0运行mapreduce之推断（speculative）执行（下）
  • 5
    MapReduce程序开发
  • 6
    初识阿里云E-MapReduce ClickHouse，到底有哪些优势？
  • 7
    Hadoop MapReduce（FlowCount） Java编程
  • 8
    MapReduce编程(三) 排序
  • 9
    MapReduce 二次排序详解
  • 10
    SQL Server优化技巧之SQL Server中的"MapReduce"
  • 1
    常用JavaScript 数组 API大全
    65
  • 2
    JavaScript循环遍历常用的7种方法以及常用的数组 API
    79
  • 3
    解释如何在 Python 中实现 Web 服务（RESTful API）。
    70
  • 4
    Java 8新特性之Lambda表达式和Stream API
    43
  • 5
    视觉智能平台常见问题之抠人脸的api如何解决
    84
  • 6
    亚马逊API接口获取商品详情的全面指南
    564
  • 7
    从API到Agent：万字长文洞悉LangChain工程化设计
    444
  • 8
    如何在Linux使用docker部署Swagger Editor并实现无公网IP远程协同编辑API文档
    123
  • 9
    虾皮Shopee根据ID取商品详情API
    131
  • 10
    谷歌发布MediaPipe LLM Inference API，28亿参数模型本地跑
    265

    • 相关课程

    更多
  • 大数据实战项目：反爬虫系统（Lua+Spark+Redis+Hadoop框架搭建）第五阶段
  • 大数据实战项目 - 反爬虫系统（Lua+Spark+Redis+Hadoop框架搭建）第六阶段
  • 大数据实战项目 - 反爬虫系统（Lua+Spark+Redis+Hadoop框架搭建）第七阶段
  • 大数据Hadoop快速入门
  • Hadoop快速入门
  • Hadoop企业优化及扩展案例

    • 相关电子书

    更多
  • 《构建Hadoop生态批流一体的实时数仓》
  • 零基础实现hadoop 迁移 MaxCompute 之 数据
  • CIO 指南:如何在SAP软件架构中使用Hadoop

    • 相关实验场景

    更多
  • 搭建Hadoop环境
  • 使用Flink实时发现最热Github项目
  • 快速体验智能体API应用
  • 基于Assistant Api的旅游助手
  • 快速体验智能体API应用
    • 下一篇
    无影云桌面