1 CxfInputFormat.java
2
3 package com.alibaba.cxf.sort;
4
5 import java.io.IOException;
6
7 import org.apache.hadoop.io.IntWritable;
8 import org.apache.hadoop.io.LongWritable;
9 import org.apache.hadoop.io.NullWritable;
10 import org.apache.hadoop.io.Text;
11 import org.apache.hadoop.mapred.FileInputFormat;
12 import org.apache.hadoop.mapred.FileSplit;
13 import org.apache.hadoop.mapred.InputSplit;
14 import org.apache.hadoop.mapred.JobConf;
15 import org.apache.hadoop.mapred.LineRecordReader;
16 import org.apache.hadoop.mapred.RecordReader;
17 import org.apache.hadoop.mapred.Reporter;
18
19 public class CxfInputFormat extends FileInputFormat<IntWritable,Text>{
20 @Override
21 public RecordReader<IntWritable, Text> getRecordReader(InputSplit split,
22 JobConf job, Reporter reporter) throws IOException {
23 return new CxfRecordReader(job, (FileSplit) split);
24 }
25 class CxfRecordReader implements RecordReader<IntWritable,Text> {
26
27 private LineRecordReader in;
28 private LongWritable junk = new LongWritable();
29 private Text line = new Text();
30 private int KEY_LENGTH = 10;
31 public CxfRecordReader(JobConf job,FileSplit split) throws IOException{
32 in = new LineRecordReader(job, split);
33 }
34 @Override
35 public void close() throws IOException {
36 in.close();
37 }
38 @Override
39 public IntWritable createKey() {
40 return new IntWritable();
41 }
42 @Override
43 public Text createValue() {
44
45 return new Text();
46 }
47 @Override
48 public long getPos() throws IOException {
49
50 return in.getPos();
51 }
52 @Override
53 public float getProgress() throws IOException {
54
55 return in.getProgress();
56 }
57 @Override
58 public boolean next(IntWritable key, Text value) throws IOException {
59 if (in.next(junk, line)) {
60 if (line.getLength() < KEY_LENGTH) {
61 key.set(Integer.parseInt(line.toString()));
62 value = new Text();
63 // value.clear();
64 } else {
65 byte[] bytes = line.getBytes();
66 key.set(Integer.parseInt(new String(bytes).substring(0, KEY_LENGTH)));
67 value = new Text();
68 }
69 return true;
70 } else {
71 return false;
72 }
73 }
74 }
75 }
76
77
78
79 SortByMapReduce.java
80
81 package com.alibaba.cxf.sort;
82
83 import java.io.IOException;
84 import java.net.URI;
85 import java.net.URISyntaxException;
86 import org.apache.hadoop.filecache.DistributedCache;
87 import org.apache.hadoop.fs.Path;
88 import org.apache.hadoop.io.IntWritable;
89 import org.apache.hadoop.io.NullWritable;
90 import org.apache.hadoop.mapred.FileInputFormat;
91 import org.apache.hadoop.mapred.FileOutputFormat;
92 import org.apache.hadoop.mapred.JobClient;
93 import org.apache.hadoop.mapred.JobConf;
94 import org.apache.hadoop.mapred.TextOutputFormat;
95 import org.apache.hadoop.mapred.lib.InputSampler;
96 import org.apache.hadoop.mapred.lib.TotalOrderPartitioner;
97 public class SortByMapReduce {
98
99 /**
100 * @param args
101 * @throws URISyntaxException
102 * @throws IOException
103 */
104 public static void main(String[] args) throws IOException, URISyntaxException {
105 runJob(args);
106 }
107
108 private static void runJob(String[] args) throws IOException, URISyntaxException {
109
110 JobConf conf = new JobConf(SortByMapReduce.class);
111
112 FileInputFormat.setInputPaths(conf, new Path(args[0]));
113 FileOutputFormat.setOutputPath(conf, new Path(args[1]));
114 conf.setJobName(”SortByMapReduce”);
115
116 conf.setInputFormat(CxfInputFormat.class);
117 conf.setOutputKeyClass(IntWritable.class);
118 conf.setOutputFormat(TextOutputFormat.class);
119 conf.setNumReduceTasks(5);
120 conf.setPartitionerClass(TotalOrderPartitioner.class);
121 InputSampler.RandomSampler<IntWritable, NullWritable> sampler =
122 new InputSampler.RandomSampler<IntWritable, NullWritable>(0.1,10000,10);
123
124 Path input = FileInputFormat.getInputPaths(conf)[0];
125 input = input.makeQualified(input.getFileSystem(conf));
126 Path partitionFile = new Path(input,”_partitions”);
127 TotalOrderPartitioner.setPartitionFile(conf, partitionFile);
128 InputSampler.writePartitionFile(conf, sampler);
129
130 URI partitionURI = new URI(partitionFile.toString() + “#_partitions”);
131 DistributedCache.addCacheFile(partitionURI, conf);
132 DistributedCache.createSymlink(conf);
133 JobClient.runJob(conf);
134 }
135 }
相关推荐
在hadoop平台上,用mapreduce编程实现大数据的词频统计
Hadoop 大数据方向 mapreduce计算中的二次排序,讲解透彻
mapreduce二次排序,年份升序,按照年份聚合,气温降序
本文针对MapReduce中的包括计数器、排序和数据集连接进行讲解。计数器包括系统计数器和自定义计数器,排序包括部分排序和文件排序,还包括对键内数据的排序。
mapreduce实现全栈排序,简单算法已经在文档中说明,想要了解的可以查看!
由于 MapReduce 中对 key 进行比较和排序,而 key 可以是任何实 现了 Writable 接口的类。 在 java 中,要实现类的大小比较可以实现 Comparable 接口并通 过重写 compareTo 方法来实现。 在 Mapreduce 中,如果需要...
对相同分区的数据,按照key进行排序(默认按照字典排序)、分组。相同key的value放在一个集合中。 (可选)分组后对数据进行归约。 注意:MapReduce中,Mapper可以单独存在,但是Reducer不能存在。
框架会对map的输出先进行排序, 然后把结果输入给reduce任务。通常作业的输入和输出都会被存储在文件系统中。 整个框架负责任务的调度和监控,以及重新执行已经失败的任务。 通常,hadoop Map/Reduce框架和分布式...
一种新颖的数据移动协议将 RDMA 与高效的归并排序算法结合使用,使基于 InfiniBand 和 10GbE 和 40GbE RoCE(RDMA over Converged Ethernet)适配器卡的 Hadoop 集群能够在加速 Hadoop 框架的服务器之间有效移动...
Hadoop Mapreduce过程shuffle过程全解析,Shuffle过程
了解map和reduce工作原理,以及排序,分组,分区设置,有详细的注释,方便查看学习,适合入门初学者练手
基于MapReduce的简单倒排索引的建立
060 MapReduce执行流程之Shuffle和排序流程以及Map端分析 061 MapReduce执行流程之Reduce端分析 062 MapReduce Shuffle过程讲解和Map Shuffle Phase讲解 063 Reduce Shuffle Phase讲解 064 源代码跟踪查看Map Task和...
基于ItemCF协同过滤、hadoop-mapreduce的商品推荐系统下载地址。 基于ItemCF的协同过滤 物品推荐系统 Collaborative filtering goods recommendation system based on ItemCF Step1.run(config, paths); // 格式化 ...
在单机环境下按多关键字对大数据排序需要较长的执行时间,为了提高按多关键字对大数据排序的效率,根据Hadoop的MapReduce模型,给出了两种基于Hadoop的多关键字排序方法。方法一在Reduce函数中使用链式基数排序算法...
hadoop实现计数器,利用mapreduce进行排序,同时可以实时跟踪整个过程以及每个节点的参与情况
本次实验,在 Hadoop 平台上,使用 MapReduce 实现了数据的全局排序。将详细阐述了实现所需环境及过程。用阿里云服务器安装, OS: Ubuntu20.04 LTS . Hadoop 支持用三种模式启动:单机模式、伪分布式模式、分布式...
Hadoop的I/O、MapReduce应用程序开发;MapReduce的工作机制;MapReduce的类型和格式;MapReduce的特性;如何构建Hadoop集群,如何管理Hadoop;Pig简介;Hbase简介;Hive简介;ZooKeeper简介;开源工具Sqoop,最后还...
04-hadoop的自定义排序实现.avi 05-mr程序中自定义分组的实现.avi 06-shuffle机制.avi 07-mr程序的组件全貌.avi 08-textinputformat对切片规划的源码分析.avi 09-倒排索引的mr实现.avi 10-多个job在同一个...
包org.dan.mr.flowsumsort MapReduce流量统计,按总流量排序 包org.dan.mr.order_pro MapReduce实现订单信息和产品信息的join逻辑 包org.dan.mr.order_pro_mapjoin MapReduce实现订单信息和产品信息的join逻辑,在...