使用MapReduce对数据文件进行切分 -

yehao0716

浏览: 22955 次
性别:
来自: 武汉

最近访客更多访客>>

日出斯图加特

luanblue

18585703436

fmq2008

博主相关

博客

微博

相册

留言

关于我

文章分类

社区版块

存档分类

使用MapReduce对数据文件进行切分

博客分类：

hadoop
linux

hadoop mapreduce partition hdfs MultipleOutPuts

有一个格式化的数据文件，用\t分割列，第2列为产品名称。现在需求把数据文件根据产品名切分为多个文件，使用MapReduce程序要如何实现？

原始文件：

[root@localhost opt]# cat aprData

1 a1 a111

2 a2 a211

3 a1 a112

4 a1 a112

5 a1 a112

6 a1 a112

7 a2 a112

8 a2 a112

9 a2 a112

10 a3 a113

思路：

1.用一个mapreduce程序找出所有产品名称：

1.1map<k2,v2>为<产品名称,null>

1.2reduce<k3,v3>为<产品名称,null>

实现：AprProduces类

[root@localhost opt]# hadoop jar apr-produces.jar /aprData /aprProduce-output

Warning: $HADOOP_HOME is deprecated.

16/05/01 15:00:12 WARN mapred.JobClient: Use GenericOptionsParser for parsing the arguments. Applications should implement Tool for the same.

16/05/01 15:00:12 INFO input.FileInputFormat: Total input paths to process : 1

16/05/01 15:00:12 INFO util.NativeCodeLoader: Loaded the native-hadoop library

16/05/01 15:00:12 WARN snappy.LoadSnappy: Snappy native library not loaded

16/05/01 15:00:13 INFO mapred.JobClient: Running job: job_201605010048_0020

16/05/01 15:00:14 INFO mapred.JobClient: map 0% reduce 0%

16/05/01 15:00:33 INFO mapred.JobClient: map 100% reduce 0%

16/05/01 15:00:45 INFO mapred.JobClient: map 100% reduce 100%

16/05/01 15:00:50 INFO mapred.JobClient: Job complete: job_201605010048_0020

16/05/01 15:00:50 INFO mapred.JobClient: Counters: 29

16/05/01 15:00:50 INFO mapred.JobClient: Map-Reduce Framework

16/05/01 15:00:50 INFO mapred.JobClient: Spilled Records=20

16/05/01 15:00:50 INFO mapred.JobClient: Map output materialized bytes=56

16/05/01 15:00:50 INFO mapred.JobClient: Reduce input records=10

16/05/01 15:00:50 INFO mapred.JobClient: Virtual memory (bytes) snapshot=3868389376

16/05/01 15:00:50 INFO mapred.JobClient: Map input records=10

16/05/01 15:00:50 INFO mapred.JobClient: SPLIT_RAW_BYTES=89

16/05/01 15:00:50 INFO mapred.JobClient: Map output bytes=30

16/05/01 15:00:50 INFO mapred.JobClient: Reduce shuffle bytes=56

16/05/01 15:00:50 INFO mapred.JobClient: Physical memory (bytes) snapshot=240697344

16/05/01 15:00:50 INFO mapred.JobClient: Reduce input groups=3

16/05/01 15:00:50 INFO mapred.JobClient: Combine output records=0

16/05/01 15:00:50 INFO mapred.JobClient: Reduce output records=3

16/05/01 15:00:50 INFO mapred.JobClient: Map output records=10

16/05/01 15:00:50 INFO mapred.JobClient: Combine input records=0

16/05/01 15:00:50 INFO mapred.JobClient: CPU time spent (ms)=1490

16/05/01 15:00:50 INFO mapred.JobClient: Total committed heap usage (bytes)=177016832

16/05/01 15:00:50 INFO mapred.JobClient: File Input Format Counters

16/05/01 15:00:50 INFO mapred.JobClient: Bytes Read=101

16/05/01 15:00:50 INFO mapred.JobClient: FileSystemCounters

16/05/01 15:00:50 INFO mapred.JobClient: HDFS_BYTES_READ=190

16/05/01 15:00:50 INFO mapred.JobClient: FILE_BYTES_WRITTEN=43049

16/05/01 15:00:50 INFO mapred.JobClient: FILE_BYTES_READ=56

16/05/01 15:00:50 INFO mapred.JobClient: HDFS_BYTES_WRITTEN=9

16/05/01 15:00:50 INFO mapred.JobClient: Job Counters

16/05/01 15:00:50 INFO mapred.JobClient: Launched map tasks=1

16/05/01 15:00:50 INFO mapred.JobClient: Launched reduce tasks=1

16/05/01 15:00:50 INFO mapred.JobClient: SLOTS_MILLIS_REDUCES=11002

16/05/01 15:00:50 INFO mapred.JobClient: Total time spent by all reduces waiting after reserving slots (ms)=0

16/05/01 15:00:50 INFO mapred.JobClient: SLOTS_MILLIS_MAPS=13561

16/05/01 15:00:50 INFO mapred.JobClient: Total time spent by all maps waiting after reserving slots (ms)=0

16/05/01 15:00:50 INFO mapred.JobClient: Data-local map tasks=1

16/05/01 15:00:50 INFO mapred.JobClient: File Output Format Counters

16/05/01 15:00:50 INFO mapred.JobClient: Bytes Written=9

[root@localhost opt]# hadoop fs -cat /aprProduce-output/part-r-00000

Warning: $HADOOP_HOME is deprecated.

2.再用一个mapreduce程序对文件进行切分：

2.1map<k2,v2>为<产品名称,line>

2.2reduce<k3,v3>为<line,null>

2.3自定义分区partition，读取第一个mapreduce程序的输出文件，组装成一个map<产品名称,index>，在partition中判断产品名称并返回下标，没有找到放在0下标中。

2.4设置taskNum(reduce的个数),taskNum应该和partition的个数一致.

3.5使用MultipleOutPuts类进行重命名输出文件，输出文件为 xxx-00001 等

实现：AprClassify类

[root@localhost opt]# hadoop jar apr-classify.jar /aprData /apr-output

Warning: $HADOOP_HOME is deprecated.

16/05/01 14:09:11 WARN mapred.JobClient: Use GenericOptionsParser for parsing the arguments. Applications should implement Tool for the same.

16/05/01 14:09:11 INFO input.FileInputFormat: Total input paths to process : 1

16/05/01 14:09:11 INFO util.NativeCodeLoader: Loaded the native-hadoop library

16/05/01 14:09:11 WARN snappy.LoadSnappy: Snappy native library not loaded

16/05/01 14:09:11 INFO mapred.JobClient: Running job: job_201605010048_0017

16/05/01 14:09:13 INFO mapred.JobClient: map 0% reduce 0%

16/05/01 14:09:29 INFO mapred.JobClient: map 100% reduce 0%

16/05/01 14:09:41 INFO mapred.JobClient: map 100% reduce 33%

16/05/01 14:09:44 INFO mapred.JobClient: map 100% reduce 66%

16/05/01 14:09:56 INFO mapred.JobClient: map 100% reduce 100%

16/05/01 14:10:01 INFO mapred.JobClient: Job complete: job_201605010048_0017

16/05/01 14:10:01 INFO mapred.JobClient: Counters: 29

16/05/01 14:10:01 INFO mapred.JobClient: Map-Reduce Framework

16/05/01 14:10:01 INFO mapred.JobClient: Spilled Records=20

16/05/01 14:10:01 INFO mapred.JobClient: Map output materialized bytes=169

16/05/01 14:10:01 INFO mapred.JobClient: Reduce input records=10

16/05/01 14:10:01 INFO mapred.JobClient: Virtual memory (bytes) snapshot=7754653696

16/05/01 14:10:01 INFO mapred.JobClient: Map input records=10

16/05/01 14:10:01 INFO mapred.JobClient: SPLIT_RAW_BYTES=89

16/05/01 14:10:01 INFO mapred.JobClient: Map output bytes=131

16/05/01 14:10:01 INFO mapred.JobClient: Reduce shuffle bytes=169

16/05/01 14:10:01 INFO mapred.JobClient: Physical memory (bytes) snapshot=387825664

16/05/01 14:10:01 INFO mapred.JobClient: Reduce input groups=3

16/05/01 14:10:01 INFO mapred.JobClient: Combine output records=0

16/05/01 14:10:01 INFO mapred.JobClient: Reduce output records=0

16/05/01 14:10:01 INFO mapred.JobClient: Map output records=10

16/05/01 14:10:01 INFO mapred.JobClient: Combine input records=0

16/05/01 14:10:01 INFO mapred.JobClient: CPU time spent (ms)=3950

16/05/01 14:10:01 INFO mapred.JobClient: Total committed heap usage (bytes)=209522688

16/05/01 14:10:01 INFO mapred.JobClient: File Input Format Counters

16/05/01 14:10:01 INFO mapred.JobClient: Bytes Read=101

16/05/01 14:10:01 INFO mapred.JobClient: FileSystemCounters

16/05/01 14:10:01 INFO mapred.JobClient: HDFS_BYTES_READ=199

16/05/01 14:10:01 INFO mapred.JobClient: FILE_BYTES_WRITTEN=86609

16/05/01 14:10:01 INFO mapred.JobClient: FILE_BYTES_READ=169

16/05/01 14:10:01 INFO mapred.JobClient: HDFS_BYTES_WRITTEN=104

16/05/01 14:10:01 INFO mapred.JobClient: Job Counters

16/05/01 14:10:01 INFO mapred.JobClient: Launched map tasks=1

16/05/01 14:10:01 INFO mapred.JobClient: Launched reduce tasks=3

16/05/01 14:10:01 INFO mapred.JobClient: SLOTS_MILLIS_REDUCES=35295

16/05/01 14:10:01 INFO mapred.JobClient: Total time spent by all reduces waiting after reserving slots (ms)=0

16/05/01 14:10:01 INFO mapred.JobClient: SLOTS_MILLIS_MAPS=13681

16/05/01 14:10:01 INFO mapred.JobClient: Total time spent by all maps waiting after reserving slots (ms)=0

16/05/01 14:10:01 INFO mapred.JobClient: Data-local map tasks=1

16/05/01 14:10:01 INFO mapred.JobClient: File Output Format Counters

16/05/01 14:10:01 INFO mapred.JobClient: Bytes Written=0

[root@localhost opt]# hadoop fs -ls /apr-output/

Warning: $HADOOP_HOME is deprecated.

Found 8 items

-rw-r--r-- 1 root supergroup 0 2016-05-01 14:09 /apr-output/_SUCCESS

drwxr-xr-x - root supergroup 0 2016-05-01 14:09 /apr-output/_logs

-rw-r--r-- 1 root supergroup 51 2016-05-01 14:09 /apr-output/a1-r-00000

-rw-r--r-- 1 root supergroup 41 2016-05-01 14:09 /apr-output/a2-r-00001

-rw-r--r-- 1 root supergroup 12 2016-05-01 14:09 /apr-output/a3-r-00002

-rw-r--r-- 1 root supergroup 0 2016-05-01 14:09 /apr-output/part-r-00000

-rw-r--r-- 1 root supergroup 0 2016-05-01 14:09 /apr-output/part-r-00001

-rw-r--r-- 1 root supergroup 0 2016-05-01 14:09 /apr-output/part-r-00002

[root@localhost opt]# hadoop fs -cat /apr-output/a1-r-00000

Warning: $HADOOP_HOME is deprecated.

1 a1 a111

3 a1 a112

4 a1 a112

5 a1 a112

6 a1 a112

[root@localhost opt]# hadoop fs -cat /apr-output/a2-r-00000

Warning: $HADOOP_HOME is deprecated.

cat: File does not exist: /apr-output/a2-r-00000

[root@localhost opt]# hadoop fs -cat /apr-output/a2-r-00001

Warning: $HADOOP_HOME is deprecated.

2 a2 a211

7 a2 a112

8 a2 a112

9 a2 a112

[root@localhost opt]# hadoop fs -cat /apr-output/a3-r-00002

Warning: $HADOOP_HOME is deprecated.

10 a3 a113

3.用hdfs对文件进行批量复制，重命名并转移产品数据文件到指定目录

实现：RenameApr类

[root@localhost opt]# hadoop fs -ls /aprProduces

Warning: $HADOOP_HOME is deprecated.

Found 3 items

-rw-r--r-- 3 yehao supergroup 51 2016-05-01 14:37 /aprProduces/a1

-rw-r--r-- 3 yehao supergroup 41 2016-05-01 14:37 /aprProduces/a2

-rw-r--r-- 3 yehao supergroup 12 2016-05-01 14:37 /aprProduces/a3

[root@localhost opt]# hadoop fs -cat /aprProduces/a1

Warning: $HADOOP_HOME is deprecated.

1 a1 a111

3 a1 a112

4 a1 a112

5 a1 a112

6 a1 a112

[root@localhost opt]# hadoop fs -cat /aprProduces/a2

Warning: $HADOOP_HOME is deprecated.

2 a2 a211

7 a2 a112

8 a2 a112

9 a2 a112

[root@localhost opt]# hadoop fs -cat /aprProduces/a3

Warning: $HADOOP_HOME is deprecated.

10 a3 a113

代码部分：

1.com.huawei.AprClassify

package com;

import java.io.IOException;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
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.mapreduce.lib.output.MultipleOutputs;


public class AprClassify {
	private static int taskNum = HdfsUtils.getMapSize();
	
	public static void main(String[] args)  throws Exception {
		Job job = new Job(new Configuration(), AprClassify.class.getSimpleName());
		job.setJarByClass(AprClassify.class);
		
		job.setMapperClass(AprClassifyMap.class);
		job.setReducerClass(AprClassifyReducer.class);
		
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(NullWritable.class);
		job.setPartitionerClass(AprClassifyPartitioner.class);
		job.setNumReduceTasks(taskNum+1);

		FileInputFormat.addInputPath(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		
		job.waitForCompletion(true);
	}
}


class AprClassifyReducer extends Reducer<Text, Text, Text, NullWritable>{
	private MultipleOutputs<Text, NullWritable> outputs; 
	
	protected void setup(Context context) throws IOException, InterruptedException {  
		outputs = new MultipleOutputs<Text, NullWritable>(context);  
	}
	
	@Override
	protected void reduce(Text k2, Iterable<Text> v2s,
			Reducer<Text, Text, Text, NullWritable>.Context context)
			throws IOException, InterruptedException {
		String st = "";
		for (Text text : v2s) {
			st += text.toString() +"\n";
		}

		Text k3 = new Text(st);
		outputs.write(k3, NullWritable.get(), k2.toString());
	}
	
	protected void cleanup(Context context) throws IOException,  
	    InterruptedException {  
		outputs.close();  
	}
}

class AprClassifyMap extends Mapper<LongWritable, Text, Text, Text>{
	Text k2 = new Text();
	
	@Override
	protected void map(LongWritable key, Text value,
			Mapper<LongWritable, Text, Text, Text>.Context context)
			throws IOException, InterruptedException {
		
		String line = value.toString();
		String[] splited = line.split("\t");
		k2.set(splited[1]);
		System.out.println(context);
		System.out.println(k2);
		System.out.println(value);
		context.write(k2, value);
	}
}

class AprClassifyPartitioner extends Partitioner<Text, Text> {

	private static Map<String, Integer> map = HdfsUtils.getMap();
	@Override
	public int getPartition(Text key, Text value, int numPartitions) {
		if(map.get(key.toString()) == null){
			return 0;
		}
		return map.get(key.toString());
	}
}

2.com.huawei.HdfsUtils

package com.huawei;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.HashMap;
import java.util.Map;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;

public class HdfsUtils {
	
	private static FileSystem fileSystem;
	private static Map<String, Integer> map;
	
	private static FileSystem getFileSystem() throws URISyntaxException, IOException {
		if(fileSystem == null){
			Configuration conf = new Configuration();
			URI uri = new URI("hdfs://192.168.1.190:9000");
			fileSystem = FileSystem.get(uri, conf);
		}
		
		return fileSystem;
	}
	
	public static int getMapSize(){
		return getMap().size();
	}
	
	public static Map<String, Integer> getMap(){
		if(map == null){
			map = new HashMap<String, Integer>();
			FSDataInputStream in;
			BufferedReader reader = null;
			try{
				fileSystem = getFileSystem();
				in = fileSystem.open(new Path("hdfs://192.168.1.190:9000/aprProduce"));
				reader = new BufferedReader(new InputStreamReader(in));
				String line = null;
				int i = 1;
				while((line = reader.readLine()) != null) {
					map.put(line, i++);
				}
			}catch(Exception e){
				 e.printStackTrace();
			}finally{
				try {
				      if(reader != null) reader.close();
				 } catch (IOException e) {
				      e.printStackTrace();
				 }
			}
		}
		
		return map;
	}
	
	public static void copyProduces(String inputPath, String outPutDir)  throws Exception{
		FileStatus[] listStatus = getFileSystem().listStatus(new Path(inputPath));
		for (FileStatus fileStatus : listStatus) {
			String name = fileStatus.getPath().getName();
			if(!fileStatus.isDir() && !StringUtils.equals(name, "_SUCCESS") && !StringUtils.startsWith(name, "part-r-")){
				FSDataInputStream openStream = fileSystem.open(fileStatus.getPath());
				IOUtils.copyBytes(openStream, fileSystem.create(new Path("/"+outPutDir+"/"+name.split("-")[0])), 1024, false);
				IOUtils.closeStream(openStream);
			}
		}
	}
}

3.com.huawei.AprProduces

package com.huawei;

import java.io.IOException;

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

/**
 * 分析文件，获得所有产品名
 * args[0] 原始文件
 * args[1] 输出文件：所有产品名
 *
 */
public class AprProduces {

	public static void main(String[] args) throws Exception {
		Job job = new Job(new Configuration(), AprProduces.class.getSimpleName());
		job.setJarByClass(AprProduces.class);
		
		job.setMapperClass(AprProducesMap.class);
		job.setReducerClass(AprProducesReducer.class);
		
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(NullWritable.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(NullWritable.class);

		FileInputFormat.addInputPath(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		
		job.waitForCompletion(true);
	}

}


class AprProducesMap extends Mapper<LongWritable, Text, Text, NullWritable>{
	Text k2 = new Text();
	
	@Override
	protected void map(LongWritable key, Text value,
			Mapper<LongWritable, Text, Text, NullWritable>.Context context)
			throws IOException, InterruptedException {
		
		String line = value.toString();
		String[] splited = line.split("\t");
		k2.set(splited[1]);//四个文件的 文件名的下标不一样，需要修改
		context.write(k2, NullWritable.get());
	}
}

class AprProducesReducer extends Reducer<Text, Text, Text, NullWritable>{
	@Override
	protected void reduce(Text k2, Iterable<Text> v2s,
			Reducer<Text, Text, Text, NullWritable>.Context context)
			throws IOException, InterruptedException {
		context.write(k2, NullWritable.get());
	}
}

4.com.huawei.RenameApr

package com.huawei;

public class RenameApr {
	public static void main(String[] args) throws Exception{
		//文件重命名
		HdfsUtils.copyProduces("/apr-output/", "aprProduce");
	}
}

Apr-classify.zip (21.9 KB)
下载次数: 6

0
顶

1
踩

分享到：

hadoop学习——arrayWritable的应用

2016-05-01 22:53
浏览 5801
评论(0)
分类:开源软件
查看更多

发表评论

您还没有登录,请您登录后再发表评论

最近访客更多访客>>

博主相关

文章分类

社区版块

存档分类

最新评论

使用MapReduce对数据文件进行切分

评论

发表评论

相关推荐

最近访客 更多访客>>

博主相关

文章分类

社区版块

存档分类

最新评论

使用MapReduce对数据文件进行切分

评论

发表评论

相关推荐

hadoop学习——arrayWritable的应用

hadoop2.5.1集群搭建：（四）配置historyServer

hadoop2.5.1集群搭建：（三）搭建yarn集群

hadoop2.5.1集群搭建：（二）搭建自动切换HA的HDFS集群

hadoop2.5.1集群搭建：（一）搭建手工切换ha的hdfs集群

CentOS编译hadoop2.5.1源码

hbase安装问题处理

hbase简介

去除hadoop启动时的警告信息

hadoop学习笔记-prc通信原理

hadoop的RPC通信

hadoop学习笔记-java操作hdfs

最近访客更多访客>>