开发者学堂课程【Hadoop 分布式计算框架 MapReduc:ReduceJoin 案例Mapper】学习笔记,与课程紧密联系,让用户快速学习知识。
课程地址:https://developer.aliyun.com/learning/course/94/detail/1547
ReduceJoin 案例 Mapper
代码示例
package com.liun.mr.reducejoin;
import java.io.IOException;
import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.lib.input.FileSplit;
public class TableMapper extends Mapper<LongWritable, Text, Text, TableBean> {
String name; TableBean tableBean = new TableBean(); Text k = new Text();
@Override protected void setup(Context context) throws IOException, InterruptedException {
// 获取文件名称 FileSplit inputSplit = (FileSplit) context.getInputSplit();
name = inputSplit.getPath().getName(); }
@Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
// 获取一行 String line = value.toString();
if (name.startsWith("order")) {// 订单表
String[] fields = line.split("\t");
tableBean.setOrder_id(fields[0]); tableBean.setP_id(fields[1]); tableBean.setAmount(Integer.parseInt(fields[2])); tableBean.setPname(""); tableBean.setFlag("order");
k.set(fields[1]);
} else {// 产品表
String[] fields = line.split("\t");
tableBean.setOrder_id(""); tableBean.setP_id(fields[0]); tableBean.setAmount(0); tableBean.setPname(fields[1]); tableBean.setFlag("pd");
k.set(fields[0]); }
// 写出 context.write(k, tableBean); } } |