使用map端连接结合分布式缓存机制实现Join算法

package com.buaa;

import java.io.BufferedReader;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.Hashtable;
import java.util.List;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.MapWritable;
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;
import org.apache.hadoop.mapreduce.lib.output.LazyOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

/** 
* @ProjectName CountViewers
* @PackageName com.buaa
* @ClassName CountViews
* @Description 通过map端连接,最后统计出 每天 每个类别 每分钟的收视人数 并按天分别输出不同的文件下
* @Author 刘吉超
* @Date 2016-06-01 16:12:08
*/
@SuppressWarnings("deprecation")
public class CountViews extends Configured implements Tool {
	/*
	 * 解析tv用户数据
	 */
	public static class ViewsMapper extends Mapper<LongWritable, Text, Text, MapWritable> {
		// 定义全局 Hashtable 对象
		private Hashtable<String, String> table = new Hashtable<String, String>();
		
		@Override
		protected void setup(Context context) throws IOException, InterruptedException {			
			// 返回本地文件路径  
            Path[] localPaths = (Path[]) context.getLocalCacheFiles();  
            if (localPaths.length == 0) {  
                throw new FileNotFoundException("Distributed cache file not found.");  
            }  
              
            // 获取本地 FileSystem实例  
            FileSystem fs = FileSystem.getLocal(context.getConfiguration()); 
            // 打开输入流  
            FSDataInputStream in = fs.open(new Path(localPaths[0].toString()));  
			
			// 创建BufferedReader读取器
			BufferedReader br = new BufferedReader(new InputStreamReader(in));
			
			String infoAddr = null;
			// 按行读取文件
			while (null != (infoAddr = br.readLine())) {
				// 将每行数据解析成数组 records
				String[] records = infoAddr.split("\t");
				/*
				 * records[0]为频道名称，records[1]为频道类别
				 * 世界地理	4
				 */
				table.put(records[0], records[1]);
			}
		}

		public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
		    /*
		     * 数据格式：机顶盒 + "@" + 日期 + "@" + 频道名称 + "@" + 开始时间+ "@" + 结束时间 
		     * 01050908200002327@2012-09-17@CCTV-1 综合@02:21:03@02:21:06	
		     */
			String[] records = value.toString().split("@"); 
			// 机顶盒
			String stbNum = records[0];
			// 日期
			String date = records[1];
			// 频道名称
			String sn = records[2];
			// 开始时间
			String s = records[3];
			// 结束时间
			String e = records[4];
			
			// 如果开始时间或结束时间为空，直接返回
			if(StringUtils.isEmpty(s) || StringUtils.isEmpty(e)){
				return ;
			}
			
			// 按每条记录的起始时间、结束时间 计算出分钟列表List
			List<String> list = ParseTime.getTimeSplit(s, e);
			
			if(list == null){
				return;
			}
			
			// 频道类别
			String channelType = StringUtils.defaultString(table.get(sn),"0");
			
			// 循环所有分钟，拆分数据记录并输出
			for (String min : list) {
				MapWritable avgnumMap = new MapWritable();
				avgnumMap.put(new Text(stbNum), new Text());
				
				/*
			     * 0@2012-09-17@02:59
			     */
				context.write(new Text(channelType + "@" + date+ "@" + min), avgnumMap);
			}
		}
	}
	
	/*
	 * 定义Combiner，合并 Mapper 输出结果 
	 */
	public static class ViewsCombiner extends Reducer<Text, MapWritable, Text, MapWritable> {
		protected void reduce(Text key, Iterable<MapWritable> values,Context context) throws IOException, InterruptedException {
			MapWritable avgnumMap = new MapWritable();
			for (MapWritable val : values) {
				// 合并相同的机顶盒号
				avgnumMap.putAll(val);
			}
			context.write(key, avgnumMap);
		}
	}
	
	/*
	 * 统计每个频道类别，每分钟的收视人数，然后按日期输出到不同文件路径下
	 */
	public static class ViewsReduce extends Reducer<Text, MapWritable, Text, Text> {
		// 声明多路径输出对象
		private MultipleOutputs<Text, Text> mos;
		
		protected void setup(Context context) throws IOException,InterruptedException {
			mos = new MultipleOutputs<Text, Text>(context);
		}
		
		protected void reduce(Text key, Iterable<MapWritable> values, Context context) throws IOException, InterruptedException {
			// 数据格式:key=channelType+date+min  value=map(stbNum)
			String[] kv = key.toString().split("@");
			// 频道类别
			String channelType = kv[0];
			// 日期
			String date = kv[1];
			// 分钟
			String min = kv[2];
			
			MapWritable avgnumMap = new MapWritable();
			for (MapWritable m : values) {
				avgnumMap.putAll(m);
			}
			
			// 按日期将数据输出到不同文件路径下
			mos.write(new Text(channelType), new Text(min + "\t" + avgnumMap.size()), date.replaceAll("-", ""));
		}

		protected void cleanup(Context context) throws IOException, InterruptedException {
			mos.close();
		}
	}
	
	@Override
	public int run(String[] arg) throws Exception {
		// 读取配置文件
		Configuration conf = new Configuration();
		
		// 判断路径是否存在，如果存在，则删除  
		Path mypath = new Path(arg[1]);
		FileSystem hdfs = mypath.getFileSystem(conf);
		if (hdfs.isDirectory(mypath)) {
			hdfs.delete(mypath, true);
		}
		
		Job job = Job.getInstance(conf,"CountViews");  
		// 设置主类
		job.setJarByClass(CountViews.class);
		
		// 输入路径
		FileInputFormat.addInputPaths(job, arg[0]+"20120917,"+arg[0]+"20120918,"+arg[0]+
				"20120919,"+arg[0]+"20120920,"+arg[0]+"20120921,"+arg[0]+"20120922,"+arg[0]+"20120923");
		// 输出路径
		FileOutputFormat.setOutputPath(job, new Path(arg[1]));
		// 去part-r-00000空文件
		LazyOutputFormat.setOutputFormatClass(job, TextOutputFormat.class);  
		
		// Mapper
		job.setMapperClass(ViewsMapper.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(MapWritable.class);
		
		// 设置Combiner
		job.setCombinerClass(ViewsCombiner.class);
		
		// Reducer
		job.setReducerClass(ViewsReduce.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		
		// 指定分布式缓存文件
		job.addCacheFile(new URI(arg[2]));
		
		//提交任务
		return job.waitForCompletion(true) ? 0 : 1;  
	}
	
	public static void main(String[] args) throws Exception {
		String[] arg = { 
				"hdfs://hadoop1:9000/buaa/tv/out/",
				"hdfs://hadoop1:9000/buaa/ctype/",
				"hdfs://hadoop1:9000/buaa/channel/channelType.csv"
			};
		int ec = ToolRunner.run(new Configuration(), new CountViews(), arg);
		System.exit(ec);
	}
}

package com.buaa;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.List;

/** 
* @ProjectName CountViewers
* @PackageName com.buaa
* @ClassName ParseTime
* @Description TODO
* @Author 刘吉超
* @Date 2016-06-01 16:11:10
*/
public class ParseTime {
	/**
	 * 提取start~end之间的分钟数
	 * 
	 * @param start
	 * @param end
	 * @return List
	 */
	public static List<String> getTimeSplit(String start, String end) {
		List<String> list = new ArrayList<String>();
		// SimpleDateFormat
		SimpleDateFormat formatDate = new SimpleDateFormat("HH:mm");
		SimpleDateFormat parseDate = new SimpleDateFormat("HH:mm:ss");
		
		/*
		 * 开始时间格式：02:21:03
		 */
		Calendar startCalendar = Calendar.getInstance();
		
		/*
		 * 结束时间格式：02:21:06
		 */
		Calendar endCalendar = Calendar.getInstance();
		
		try {
			startCalendar.setTime(parseDate.parse(start));
			endCalendar.setTime(parseDate.parse(end));
		} catch (ParseException e1) {
			return null;
		}
		
		while (startCalendar.compareTo(endCalendar) <= 0) {
			list.add(formatDate.format(startCalendar.getTime()));
			startCalendar.add(Calendar.MINUTE, 1);
		}
		
		return list;
	}
	
	public static void main(String[] args) {
		String start = "12:59:24";
		String end = "13:03:45";
		List<String> list1  = getTimeSplit(start, end);
		for(String st1 : list1){
			System.out.println(st1);
		}
	}
}