安科网

Hadoop之MapReduce学习(三)之ip去重、MaxScore示例、TotalScoreMapper示例

香帅

香帅

2020-04-15

关注 关注

一、IP去重示例

数据文件：

192.168.10.111
192.168.10.111
10.32.100.111
192.168.21.111
192.168.10.112
192.168.10.111
192.168.11.111
192.168.12.112
192.168.11.111
IPMapper：
package com.blb.ip;

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

import java.io.IOException;

public class IPMapper
        extends Mapper<LongWritable, Text, Text, NullWritable> {

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        // 每一行数据就是一个ip
        context.write(value, NullWritable.get());
    }
}
IPReducer：
package com.blb.ip;

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

import java.io.IOException;

public class IPReducer
        extends Reducer<Text, NullWritable, Text, NullWritable> {

    // key：ip
    // values：null,null,null...
    @Override
    protected void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {

        context.write(key, NullWritable.get());

    }
}
IPDriver：
package com.blb.ip;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
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;

public class IPDriver {

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);

        job.setJarByClass(IPDriver.class);
        job.setMapperClass(IPMapper.class);
        job.setReducerClass(IPReducer.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);

        FileInputFormat.addInputPath(job,
                new Path("hdfs://192.168.10.131:9000/txt/ip.txt"));

        FileOutputFormat.setOutputPath(job,
                new Path("hdfs://192.168.10.131:9000/result/ip"));

        job.waitForCompletion(true);
    }
}

二、axScore示例

数据文件：

张三 684
李四 312
王五 721
赵六 548
田七 470
王八 668
陈九 340

MaxScoreMapper：

package com.blb.maxscore;

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;

public class MaxScoreMapper
extends Mapper<LongWritable, Text, Text, IntWritable> {

@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

String[] arr = value.toString().split(" ");
context.write(new Text(arr[0]),
new IntWritable(Integer.parseInt(arr[1])));
}
}

MaxScoreReducer：

package com.blb.maxscore;

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

import java.io.IOException;

public class MaxScoreReducer
extends Reducer<Text, IntWritable, Text, IntWritable> {

@Override
protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
// 在MapReduce中，为了节省内存，减少对象的创建和销毁，采取了地址复用的机制
// 即在迭代过程中，对象只创建一次
IntWritable max = new IntWritable(0);
// IntWritable max = new IntWritable(0);
// IntWritable value = new IntWritable();
// values:684 312 721 548...
// value.set(684);
// value.get() > max.get() --- 684 > 0 --- true
// max = value; 此时max和value都是引用类型，给的是地址
// 也就意味着max和value的地址一样
// value.set(312); 此时max的值也是312
// value.get() > max.get() --- 312 > 312 --- false
// 继续迭代，max和value的地址始终一致
// max的值就是迭代的最后一个值
for (IntWritable value : values) {
if (value.get() > max.get())
// max = value;
max.set(value.get());
}
context.write(key, max);
}
}

MaxScoreDriver：

package com.blb.maxscore;

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;

public class MaxScoreDriver {

public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

Configuration conf = new Configuration();
Job job = Job.getInstance(conf);

job.setJarByClass(MaxScoreDriver.class);
job.setMapperClass(MaxScoreMapper.class);
job.setReducerClass(MaxScoreReducer.class);

job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);

FileInputFormat.addInputPath(job,
new Path("hdfs://192.168.10.131:9000/txt/score2.txt"));

FileOutputFormat.setOutputPath(job,
new Path("hdfs://192.168.10.131:9000/result/maxscore2"));

job.waitForCompletion(true);
}
}

 三、TotalScoreMapper示例

数据文件：

张三 78
李四 66
王五 73
张三 88
田七 75
张三 65
陈九 90
李四 67
王五 78

IPMapper：

package com.blb.maxscore;

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;

public class MaxScoreMapper
extends Mapper<LongWritable, Text, Text, IntWritable> {

@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

String[] arr = value.toString().split(" ");
context.write(new Text(arr[0]),
new IntWritable(Integer.parseInt(arr[1])));
}
}

TotalScoreMapper：

package com.blb.totalscore;

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;

public class TotalScoreMapper
extends Mapper<LongWritable, Text, Text, IntWritable> {

@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
// 拆分名字和分数
String[] arr = value.toString().split(" ");
context.write(new Text(arr[0]),
new IntWritable(Integer.parseInt(arr[1])));
}
}

TotalScoreReducer：

package com.blb.totalscore;

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

import java.io.IOException;

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

}

TotalScoreDriver：

package com.blb.totalscore;

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;

public class TotalScoreDriver {
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

Configuration conf = new Configuration();
Job job = Job.getInstance(conf);

job.setJarByClass(TotalScoreDriver.class);
job.setMapperClass(TotalScoreMapper.class);
job.setReducerClass(TotalScoreReducer.class);

job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);

FileInputFormat.addInputPath(job,
new Path("hdfs://192.168.10.131:9000/txt/score2/"));

FileOutputFormat.setOutputPath(job,
new Path("hdfs://192.168.10.131:9000/result/totalscore"));

job.waitForCompletion(true);
}
}

 

mapreduce hadoop apache

香帅

香帅

0 关注 0 粉丝 0 动态

关注 关注

相关推荐

hadoop框架三大组件hdfs、mapreduce、yarn 内容

1）分布式的运算程序往往需要分成至少2个阶段。2）第一个阶段的MapTask并发实例，完全并行运行，互不相干。3）第二个阶段的ReduceTask并发实例互不相干，但是他们的数据依赖于上一个阶段的所有MapTask并发实例的输出。4）MapReduce编程

tomli 2020-07-26

JStorm介绍

JStorm是一个类似于Hadoop MapReduce的系统，用户按照指定的接口实现一个任务，然后将这个任务交给JStorm系统，JStorm将这个任务跑起来，并按7*24小时运行。如果中间一个worker发生了意外故障，调度器立即分配一个新的worke

csdnhadoop 1评论 2020-07-04

mapreduce求topN

利用TreeSet排序,该方式利用小顶堆和集合重复原理的方式 , 每过来一个数据 , 跟堆顶数据进行比较 , 如果比最小的大 , 则将过来的数据替换堆顶元素 , 否则直接跳过数据 . 以此对数据进行排序 .public static class MapTa

ErixHao 2020-05-16

HBase与MapReduce交互

飞鸿踏雪0 2020-05-07

分布式计算框架——MapReduce

DAG计算——多个作业存在依赖关系，后一个的输入是前一个的输出，构成有向无环图DAG；Map、Reduce阶段的中间环节，负责执行Partition（分区）、Sort（排序）、Spill（溢写）、Merge（合并）、抓取等工作。Partition编号 =

zhangll00 2020-05-07

HBase和MapReduce

Apache MapReduce 是一个用于分析大量数据的软件框架。MapReduce 本身超出了本文档的范围。MapReduce 版本2现在是 YARN 的一部分。本章讨论在 HBase 中对数据使用 MapReduce 时需要采取的具体配置步骤。另外，

strongyoung 2020-04-11

MapReduce(分布式计算)_01

2-MapReduce的构思和框架结构。5-MapReduce案例-wordcount-准备工作

zhangll00 2020-03-27

MapReduce编程模型

1.环境搭建起来 伪分布式2.MapReduce是基于HDFS之上的 MapReduce计算的内容应该是HDFS上的文件/文件夹3.计算=>统计=>Mapper阶段=>Reduce阶段　　MapReduce的编程模型4.Mapper

sujins 2020-03-06

bitmap、Trie、数据库索引、倒排索引、外排序、Mapreduce

问题给40亿个不重复的unsigned int的整数，没排过序的，然后再给一个数，如何快速判断这个数是否在那40亿个数当中？方案1：用位图/Bitmap的方法，申请512M的内存，一个bit位代表一个unsigned int值。读入40亿个数，设置相应的b

lixiaotao 2020-03-05

MapReduce的使用

MapReduce基于yarn组件，想要做MapReduce就必须先开启hdfs和yarn。将common、hdfs、yarn、MapReduce文件夹下的所有jar包及其依赖包导入到项目中。Mapper阶段：Map必须得继承Mapper类，并且重写map

tomli 2020-03-05

Hadoop基础---MapReduce对数据进行排序

-1 : 1; //返回值为-1，则排在前面。return "" + up_flow + "\t" + down_flow + "\t"+ sum_flow;public class Res

WeiHHH 2020-02-23

Hadoop——MapReduce介绍

MapReduce是一种编程模型，用于大规模数据集的并行运算。概念"Map（映射）"和"Reduce（归约）"，是它们的主要思想。　　首先2T的文件，你放在单机上可能存不下来，如果再他多一点呢？比如几千个文件，几十个

zzjmay 2020-02-23

Hadoop架构及集群

HDFS为海量的数据提供了存储，而MapReduce则为海量的数据提供了计算，特点是：高可靠性，高扩展性，高效性，高容错性。

JavaWDB 2020-02-13

MapReduce编程实战（2）-词频统计结果存入mysql数据库

通过实现MapReduce计算结果保存到MySql数据库过程，掌握多种方式保存计算结果的技术，加深了对MapReduce的理解；创建maven项目，项目名称hdfs，这里不再说明。红色部分为增加内容：

大白配小猪 2020-02-09

MapReduce和Hive学习文档链接学习顺序

18、《sqoop的使用之导入到hive和mysql》。21、《Hive中利用日志数据学习分析思路》

IT智囊 2020-02-02

Hadoop学习之路(7)MapReduce自定义排序

MapReduce中，根据key进行分区、排序、分组MapReduce会按照基本类型对应的key进行排序，如int类型的IntWritable，long类型的LongWritable,Text类型，默认升序排序为什么要自定义排序规则？现有需求，需要自定义k

IT智囊 2020-01-01

python| MongoDB聚合(count、distinct、group、MapReduce)

增加查询条件会使count查询变慢。initial：每一组reduce函数调用时作为第二个参数传递给reduce函数的初始文档，每一组的所有成员都会使用这个累加器，所以改变会被保留住。group和MapReduce对比示例：查询相同年龄人的名字。

yogoma 2020-01-10

Hadoop

Hadoop是一个由Apache基金会所开发的分布式系统基础架构。充分利用集群的威力进行高速运算和存储。Hadoop实现了一个分布式文件系统，简称HDFS。HDFS放宽了POSIX的要求，可以以流的形式访问文件系统中的数据。Hadoop的框架最核心的设计就

sujins 2020-01-09

hadoop自带性能测试

清空数据hadoop jar hadoop-mapreduce-client-jobclient-2.6.4.jar -cleanhadoop fs -ls /benchmarks/. 写性能测试hadoop jar hadoop-mapreduce-c

飞鸿踏雪0 2019-12-29

Hadoop学习之路(6)MapReduce自定义分区实现

MapReduce自带的分区器是HashPartitioner原理：先对map输出的key求hash值，再模上reduce task个数，根据结果，决定此输出kv对，被匹配的reduce任务取走。自定义分分区需要继承Partitioner，复写getpar

Elmo 2019-12-29
香帅

香帅

W3CSchool教程
HTML 教程
CSS 教程
Bootstrap 教程
Javascript 教程
jQuery 教程
后端教程
C 教程
Java 教程
PHP 教程
Python 教程
Go 教程
移动开发
Android 教程
Swift 教程
Kotlin 教程
jQuery Mobile 教程
ionic 教程
关于我们
新闻动态
联系方式
招聘英才
安科实验室
帮助与反馈

安科网(Ancii)，中国第一极客网

Copyright © 2013 - 2019 Ancii.com

京ICP备18063983号 京公网安备11010802014868号