MapReduce快速入门

如何理解map、reduce计算模型

Mapreudce程序运行演示

Mapreduce编程规范及示例编写

Mapreduce程序运行模式及debug方法

MapReduce高级特性

Mapreduce程序的核心机制

MapReduce的序列化框架

MapReduce的排序实现

MapReduce的分区机制及自定义

Mapreduce的数据压缩

Mapreduce与yarn的结合

Mapreduce编程案例

Mapreduce 参数优化

//序列化，将对象的字段信息写入输出流

@Override

public void write(DataOutput out) throws IOException {

out.writeLong(upflow);

out.writeLong(downflow);

out.writeLong(sumflow);

}

//反序列化，从输入流中读取各个字段信息

@Override

public void readFields(DataInput in) throws IOException {

upflow = in.readLong();

downflow = in.readLong();

sumflow = in.readLong();

}

         @Override

         public int getPartition(Text key, FlowBean value, int numPartitions) {

                   String prefix = key.toString().substring(0,3);

                   Integer partNum = pmap.get(prefix);

                   return (partNum==null?4:partNum);

         }

job.setNumReduceTasks(5);

第一步

map

读一行   A:B,C,D,F,E,O

输出    <B,A><C,A><D,A><F,A><E,A><O,A>

在读一行   B:A,C,E,K

输出   <A,B><C,B><E,B><K,B>

REDUCE

拿到的数据比如<C,A><C,B><C,E><C,F><C,G>......

输出：

<A-B,C>

<A-E,C>

<A-F,C>

<A-G,C>

<B-E,C>

<B-F,C>.....

第二步

map

读入一行<A-B,C>

直接输出<A-B,C>

reduce

读入数据  <A-B,C><A-B,F><A-B,G>.......

输出： A-B  C,F,G,.....

public class WholeFileInputFormat extends

FileInputFormat<NullWritable, BytesWritable> {

//设置每个小文件不可分片,保证一个小文件生成一个key-value键值对

@Override

protected boolean isSplitable(JobContext context, Path file) {

return false;

}

@Override

public RecordReader<NullWritable, BytesWritable> createRecordReader(

InputSplit split, TaskAttemptContext context) throws IOException,

InterruptedException {

WholeFileRecordReader reader = new WholeFileRecordReader();

reader.initialize(split, context);

return reader;

}

}

class WholeFileRecordReader extends RecordReader<NullWritable, BytesWritable> {

private FileSplit fileSplit;

private Configuration conf;

private BytesWritable value = new BytesWritable();

private boolean processed = false;

@Override

public void initialize(InputSplit split, TaskAttemptContext context)

throws IOException, InterruptedException {

this.fileSplit = (FileSplit) split;

this.conf = context.getConfiguration();

}

@Override

public boolean nextKeyValue() throws IOException, InterruptedException {

if (!processed) {

byte[] contents = new byte[(int) fileSplit.getLength()];

Path file = fileSplit.getPath();

FileSystem fs = file.getFileSystem(conf);

FSDataInputStream in = null;

try {

in = fs.open(file);

IOUtils.readFully(in, contents, 0, contents.length);

value.set(contents, 0, contents.length);

} finally {

IOUtils.closeStream(in);

}

processed = true;

return true;

}

return false;

}

@Override

public NullWritable getCurrentKey() throws IOException,

InterruptedException {

return NullWritable.get();

}

@Override

public BytesWritable getCurrentValue() throws IOException,

InterruptedException {

return value;

}

@Override

public float getProgress() throws IOException {

return processed ? 1.0f : 0.0f;

}

@Override

public void close() throws IOException {

// do nothing

}

}

public class SmallFilesToSequenceFileConverter extends Configured implements

Tool {

static class SequenceFileMapper extends

Mapper<NullWritable, BytesWritable, Text, BytesWritable> {

private Text filenameKey;

@Override

protected void setup(Context context) throws IOException,

InterruptedException {

InputSplit split = context.getInputSplit();

Path path = ((FileSplit) split).getPath();

filenameKey = new Text(path.toString());

}

@Override

protected void map(NullWritable key, BytesWritable value,

Context context) throws IOException, InterruptedException {

context.write(filenameKey, value);

}

}

@Override

public int run(String[] args) throws Exception {

Configuration conf = new Configuration();

System.setProperty("HADOOP_USER_NAME", "hdfs");

String[] otherArgs = new GenericOptionsParser(conf, args)

.getRemainingArgs();

if (otherArgs.length != 2) {

System.err.println("Usage: combinefiles <in> <out>");

System.exit(2);

}

Job job = Job.getInstance(conf,"combine small files to sequencefile");

//               job.setInputFormatClass(WholeFileInputFormat.class);

job.setOutputFormatClass(SequenceFileOutputFormat.class);

job.setOutputKeyClass(Text.class);

job.setOutputValueClass(BytesWritable.class);

job.setMapperClass(SequenceFileMapper.class);

return job.waitForCompletion(true) ? 0 : 1;

}

public static void main(String[] args) throws Exception {

int exitCode = ToolRunner.run(new SmallFilesToSequenceFileConverter(),

args);

System.exit(exitCode);

}

}

public class DBLoader {

public static void dbLoader(HashMap<String, String> ruleMap) {

Connection conn = null;

Statement st = null;

ResultSet res = null;

try {

Class.forName("com.mysql.jdbc.Driver");

conn = DriverManager.getConnection("jdbc:mysql://hdp-node01:3306/urlknowledge", "root", "root");

st = conn.createStatement();

res = st.executeQuery("select url,content from urlcontent");

while (res.next()) {

ruleMap.put(res.getString(1), res.getString(2));

}

} catch (Exception e) {

e.printStackTrace();

} finally {

try{

if(res!=null){

res.close();

}

if(st!=null){

st.close();

}

if(conn!=null){

conn.close();

}

}catch(Exception e){

e.printStackTrace();

}

}

}

public static void main(String[] args) {

DBLoader db = new DBLoader();

HashMap<String, String> map = new HashMap<String,String>();

db.dbLoader(map);

System.out.println(map.size());

}

}

public class LogEnhancerOutputFormat extends FileOutputFormat<Text, NullWritable>{

@Override

public RecordWriter<Text, NullWritable> getRecordWriter(TaskAttemptContext context) throws IOException, InterruptedException {

FileSystem fs = FileSystem.get(context.getConfiguration());

Path enhancePath = new Path("hdfs://hdp-node01:9000/flow/enhancelog/enhanced.log");

Path toCrawlPath = new Path("hdfs://hdp-node01:9000/flow/tocrawl/tocrawl.log");

FSDataOutputStream enhanceOut = fs.create(enhancePath);

FSDataOutputStream toCrawlOut = fs.create(toCrawlPath);

return new MyRecordWriter(enhanceOut,toCrawlOut);

}

static class MyRecordWriter extends RecordWriter<Text, NullWritable>{

FSDataOutputStream enhanceOut = null;

FSDataOutputStream toCrawlOut = null;

public MyRecordWriter(FSDataOutputStream enhanceOut, FSDataOutputStream toCrawlOut) {

this.enhanceOut = enhanceOut;

this.toCrawlOut = toCrawlOut;

}

@Override

public void write(Text key, NullWritable value) throws IOException, InterruptedException {

//有了数据，你来负责写到目的地  —— hdfs

//判断，进来内容如果是带tocrawl的，就往待爬清单输出流中写 toCrawlOut

if(key.toString().contains("tocrawl")){

toCrawlOut.write(key.toString().getBytes());

}else{

enhanceOut.write(key.toString().getBytes());

}

}

@Override

public void close(TaskAttemptContext context) throws IOException, InterruptedException {

if(toCrawlOut!=null){

toCrawlOut.close();

}

if(enhanceOut!=null){

enhanceOut.close();

}

}

}

}

/**

* 这个程序是对每个小时不断产生的用户上网记录日志进行增强(将日志中的url所指向的网页内容分析结果信息追加到每一行原始日志后面)

*

* @author

*

*/

public class LogEnhancer {

static class LogEnhancerMapper extends Mapper<LongWritable, Text, Text, NullWritable> {

HashMap<String, String> knowledgeMap = new HashMap<String, String>();

/**

* maptask在初始化时会先调用setup方法一次 利用这个机制，将外部的知识库加载到maptask执行的机器内存中

*/

@Override

protected void setup(org.apache.hadoop.mapreduce.Mapper.Context context) throws IOException, InterruptedException {

DBLoader.dbLoader(knowledgeMap);

}

@Override

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

String line = value.toString();

String[] fields = StringUtils.split(line, "\t");

try {

String url = fields[26];

// 对这一行日志中的url去知识库中查找内容分析信息

String content = knowledgeMap.get(url);

// 根据内容信息匹配的结果，来构造两种输出结果

String result = "";

if (null == content) {

// 输往待爬清单的内容

result = url + "\t" + "tocrawl\n";

} else {

// 输往增强日志的内容

result = line + "\t" + content + "\n";

}

context.write(new Text(result), NullWritable.get());

} catch (Exception e) {

}

}

}

public static void main(String[] args) throws Exception {

Configuration conf = new Configuration();

Job job = Job.getInstance(conf);

job.setJarByClass(LogEnhancer.class);

job.setMapperClass(LogEnhancerMapper.class);

job.setOutputKeyClass(Text.class);

job.setOutputValueClass(NullWritable.class);

// 要将自定义的输出格式组件设置到job中

job.setOutputFormatClass(LogEnhancerOutputFormat.class);

FileInputFormat.setInputPaths(job, new Path(args[0]));

// 虽然我们自定义了outputformat，但是因为我们的outputformat继承自fileoutputformat

// 而fileoutputformat要输出一个_SUCCESS文件，所以，在这还得指定一个输出目录

FileOutputFormat.setOutputPath(job, new Path(args[1]));

job.waitForCompletion(true);

System.exit(0);

}

}

订单id

商品id

成交金额

Order_0000001

Pdt_01

222.8

Order_0000001

Pdt_05

25.8

Order_0000002

Pdt_03

522.8

Order_0000002

Pdt_04

122.4

Order_0000002

Pdt_05

722.4

Order_0000003

Pdt_01

222.8

/**

* 用于控制shuffle过程中reduce端对kv对的聚合逻辑

* @author duanhaitao@itcast.cn

*

*/

public class ItemidGroupingComparator extends WritableComparator {

protected ItemidGroupingComparator() {

super(OrderBean.class, true);

}

@Override

public int compare(WritableComparable a, WritableComparable b) {

OrderBean abean = (OrderBean) a;

OrderBean bbean = (OrderBean) b;

//将item_id相同的bean都视为相同，从而聚合为一组

return abean.getItemid().compareTo(bbean.getItemid());

}

}

/**

* 订单信息bean，实现hadoop的序列化机制

* @author duanhaitao@itcast.cn

*

*/

public class OrderBean implements WritableComparable<OrderBean>{

private Text itemid;

private DoubleWritable amount;

public OrderBean() {

}

public OrderBean(Text itemid, DoubleWritable amount) {

set(itemid, amount);

}

public void set(Text itemid, DoubleWritable amount) {

this.itemid = itemid;

this.amount = amount;

}

public Text getItemid() {

return itemid;

}

public DoubleWritable getAmount() {

return amount;

}

@Override

public int compareTo(OrderBean o) {

int cmp = this.itemid.compareTo(o.getItemid());

if (cmp == 0) {

cmp = -this.amount.compareTo(o.getAmount());

}

return cmp;

}

@Override

public void write(DataOutput out) throws IOException {

out.writeUTF(itemid.toString());

out.writeDouble(amount.get());

}

@Override

public void readFields(DataInput in) throws IOException {

String readUTF = in.readUTF();

double readDouble = in.readDouble();

this.itemid = new Text(readUTF);

this.amount= new DoubleWritable(readDouble);

}

@Override

public String toString() {

return itemid.toString() + "\t" + amount.get();

}

}

/**

* 利用secondarysort机制输出每种item订单金额最大的记录

* @author duanhaitao@itcast.cn

*

*/

public class SecondarySort {

static class SecondarySortMapper extends Mapper<LongWritable, Text, OrderBean, NullWritable>{

OrderBean bean = new OrderBean();

@Override

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

String line = value.toString();

String[] fields = StringUtils.split(line, "\t");

bean.set(new Text(fields[0]), new DoubleWritable(Double.parseDouble(fields[1])));

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

}

}

static class SecondarySortReducer extends Reducer<OrderBean, NullWritable, OrderBean, NullWritable>{

//在设置了groupingcomparator以后，这里收到的kv数据 就是：  <1001 87.6>,null  <1001 76.5>,null  ....

//此时，reduce方法中的参数key就是上述kv组中的第一个kv的key：<1001 87.6>

//要输出同一个item的所有订单中最大金额的那一个，就只要输出这个key

@Override

protected void reduce(OrderBean key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {

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

}

}

public static void main(String[] args) throws Exception {

Configuration conf = new Configuration();

Job job = Job.getInstance(conf);

job.setJarByClass(SecondarySort.class);

job.setMapperClass(SecondarySortMapper.class);

job.setReducerClass(SecondarySortReducer.class);

job.setOutputKeyClass(OrderBean.class);

job.setOutputValueClass(NullWritable.class);

FileInputFormat.setInputPaths(job, new Path(args[0]));

FileOutputFormat.setOutputPath(job, new Path(args[1]));

//指定shuffle所使用的GroupingComparator类

job.setGroupingComparatorClass(ItemidGroupingComparator.class);

//指定shuffle所使用的partitioner类

job.setPartitionerClass(ItemIdPartitioner.class);

job.setNumReduceTasks(3);

job.waitForCompletion(true);

}

}

public class TestDistributedCache {

static class TestDistributedCacheMapper extends Mapper<LongWritable, Text, Text, Text>{

FileReader in = null;

BufferedReader reader = null;

HashMap<String,String> b_tab = new HashMap<String, String>();

String localpath =null;

String uirpath = null;

//是在map任务初始化的时候调用一次

@Override

protected void setup(Context context) throws IOException, InterruptedException {

//通过这几句代码可以获取到cache file的本地绝对路径，测试验证用

Path[] files = context.getLocalCacheFiles();

localpath = files[0].toString();

URI[] cacheFiles = context.getCacheFiles();

//缓存文件的用法——直接用本地IO来读取

//这里读的数据是map task所在机器本地工作目录中的一个小文件

in = new FileReader("b.txt");

reader =new BufferedReader(in);

String line =null;

while(null!=(line=reader.readLine())){

String[] fields = line.split(",");

b_tab.put(fields[0],fields[1]);

}

IOUtils.closeStream(reader);

IOUtils.closeStream(in);

}

@Override

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

//这里读的是这个map task所负责的那一个切片数据（在hdfs上）

String[] fields = value.toString().split("\t");

String a_itemid = fields[0];

String a_amount = fields[1];

String b_name = b_tab.get(a_itemid);

// 输出结果  1001      98.9 banan

context.write(new Text(a_itemid), new Text(a_amount + "\t" + ":" + localpath + "\t" +b_name ));

}

}

public static void main(String[] args) throws Exception {

Configuration conf = new Configuration();

Job job = Job.getInstance(conf);

job.setJarByClass(TestDistributedCache.class);

job.setMapperClass(TestDistributedCacheMapper.class);

job.setOutputKeyClass(Text.class);

job.setOutputValueClass(LongWritable.class);

//这里是我们正常的需要处理的数据所在路径

FileInputFormat.setInputPaths(job, new Path(args[0]));

FileOutputFormat.setOutputPath(job, new Path(args[1]));

//不需要reducer

job.setNumReduceTasks(0);

//分发一个文件到task进程的工作目录

job.addCacheFile(new URI("hdfs://hadoop-server01:9000/cachefile/b.txt"));

//分发一个归档文件到task进程的工作目录

//               job.addArchiveToClassPath(archive);

//分发jar包到task节点的classpath下

//               job.addFileToClassPath(jarfile);

job.waitForCompletion(true);

}

}

public class MultiOutputs {

//通过枚举形式定义自定义计数器

enum MyCounter{MALFORORMED,NORMAL}

static class CommaMapper extends Mapper<LongWritable, Text, Text, LongWritable> {

@Override

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

String[] words = value.toString().split(",");

for (String word : words) {

context.write(new Text(word), new LongWritable(1));

}

//对枚举定义的自定义计数器加1

context.getCounter(MyCounter.MALFORORMED).increment(1);

//通过动态设置自定义计数器加1

context.getCounter("counterGroupa", "countera").increment(1);

}

}

ControlledJob cJob1 = new ControlledJob(job1.getConfiguration());

ControlledJob cJob2 = new ControlledJob(job2.getConfiguration());

ControlledJob cJob3 = new ControlledJob(job3.getConfiguration());

cJob1.setJob(job1);

cJob2.setJob(job2);

cJob3.setJob(job3);

// 设置作业依赖关系

cJob2.addDependingJob(cJob1);

cJob3.addDependingJob(cJob2);

JobControl jobControl = new JobControl("RecommendationJob");

jobControl.addJob(cJob1);

jobControl.addJob(cJob2);

jobControl.addJob(cJob3);

// 新建一个线程来运行已加入JobControl中的作业，开始进程并等待结束

Thread jobControlThread = new Thread(jobControl);

jobControlThread.start();

while (!jobControl.allFinished()) {

Thread.sleep(500);

}

jobControl.stop();

return 0;