MRUnitTest (Mapreduce local mode implementation 0.27.7-public-java8 API)

java.lang.Object
- com.aliyun.odps.mapred.unittest.MRUnitTest

public class MRUnitTest
extends Object

MapReduce 单元测试基类，方便用户编写 MapReduce 作业的单元测试用例.

MRUnitTest 提供两个主要的方法用于编写 Mapper 和 Reducer 的单元测试用例：

runMapper(JobConf, MapUTContext) - 执行 Mapper 过程
runReducer(JobConf, ReduceUTContext) - 执行 Reducer 过程

另外，提供记录对象的创建、从文件读取、持久化到文件以及记录比较的一些常用方法。

代码示例：

 package com.aliyun.odps.mapred.example;
 
 import java.io.IOException;
 import java.util.Iterator;
 import java.util.List;
 
 import junit.framework.Assert;
 
 import org.junit.Test;
 
 import com.aliyun.odps.data.Record;
 import com.aliyun.odps.data.TableInfo;
 import com.aliyun.odps.io.Text;
 import com.aliyun.odps.mapred.MapperBase;
 import com.aliyun.odps.mapred.ReducerBase;
 import com.aliyun.odps.mapred.conf.JobConf;
 import com.aliyun.odps.mapred.utils.InputUtils;
 import com.aliyun.odps.mapred.utils.OutputUtils;
 import com.aliyun.odps.mapred.utils.SchemaUtils;
 import com.aliyun.odps.mapred.unittest.*;
 
 public class WordCountTest extends MRUnitTest {
 
   private final static String INPUT_SCHEMA = "a:string,b:string";
   private final static String OUTPUT_SCHEMA = "k:string,v:bigint";
 
   @Test
   public void TestMapReduce() throws IOException, ClassNotFoundException,
       InterruptedException {
 
     JobConf job = new JobConf();
     
     job.setMapperClass(TokenizerMapper.class);
     job.setCombinerClass(SumCombiner.class);
     job.setReducerClass(SumReducer.class);
     
     job.setMapOutputKeySchema(SchemaUtils.fromString("key:string"));
     job.setMapOutputValueSchema(SchemaUtils.fromString("value:bigint"));
     
     InputUtils.addTable(TableInfo.builder().tableName("wc_in").build(), job);
     OutputUtils.addTable(TableInfo.builder().tableName("wc_out").build(), job);
 
     // 准备 Mapper 的 MapUTContext 并指定输入输出 schema
     MapUTContext mapContext = new MapUTContext();
     mapContext.setInputSchema(INPUT_SCHEMA);
     mapContext.setOutputSchema(OUTPUT_SCHEMA);
 
     // 准备 Mapper 的输入记录
     Record record = mapContext.createInputRecord();
     record.set(new Text[] {new Text("hello"), new Text("c")});
     mapContext.addInputRecord(record);
     record = mapContext.createInputRecord();
     record.set(new Text[] {new Text("hello"), new Text("java")});
     mapContext.addInputRecord(record);
 
     // 开始执行 Mapper 过程
     TaskOutput mapOutput = runMapper(job, mapContext);
 
     // 验证 Mapper的执行结果
     List<KeyValue<Record, Record>> kvs = mapOutput.getOutputKeyValues();
     Assert.assertEquals(3, kvs.size());
     Assert.assertEquals(new KeyValue<String, Long>(new String("c"), new Long(1)),
         new KeyValue<String, Long>((String) (kvs.get(0).getKey().get(0)), (Long) (kvs.get(0)
         .getValue().get(0))));
     Assert.assertEquals(new KeyValue<String, Long>(new String("hello"), new Long(2)),
         new KeyValue<String, Long>((String) (kvs.get(1).getKey().get(0)), (Long) (kvs.get(1)
         .getValue().get(0))));
     Assert.assertEquals(new KeyValue<String, Long>(new String("java"), new Long(1)),
         new KeyValue<String, Long>((String) (kvs.get(2).getKey().get(0)), (Long) (kvs.get(2)
         .getValue().get(0))));
 
     // 准备 Reducer 的输出 schema 和 输入键值对（Key/Value）
     ReduceUTContext reduceContext = new ReduceUTContext();
     reduceContext.setOutputSchema(OUTPUT_SCHEMA);
 
     // 准备 Reducer 的输入键值对，这里把 Mapper 的结果作为 Reducer 的输入
     reduceContext.addInputKeyValues(mapOutput);
 
     // 执行 Reducer 过程
     ReduceOutput output = runReducer(job, reduceContext);
 
     // 验证 Reducer 的执行结果
     List<Record> records = output.getOutputRecords();
     Assert.assertEquals(3, records.size());
     Assert.assertEquals(new String("c"), records.get(0).get("k"));
     Assert.assertEquals(new Long(1), records.get(0).get("v"));
     Assert.assertEquals(new String("hello"), records.get(1).get("k"));
     Assert.assertEquals(new Long(2), records.get(1).get("v"));
     Assert.assertEquals(new String("java"), records.get(2).get("k"));
     Assert.assertEquals(new Long(1), records.get(2).get("v"));
   }
 }

Constructor Summary

Constructors
Constructor and Description

MRUnitTest()

Constructors
Constructor and Description
`MRUnitTest()`

Method Summary

All Methods Static Methods Instance Methods Concrete Methods
Modifier and Type	Method and Description
`static Record`	`createRecord(String schema)` 给定 schema 创建记录对象（`Record`）.
`static boolean`	`equalRecords(File dir, List<Record> records, boolean sort)` 比较给定本地目录下的记录集合和一个记录列表是否相同（相等），记录比较见 `RecordComparator`.
`static boolean`	`equalRecords(List<Record> records1, List<Record> records2, boolean sort)` 比较两个记录列表是否相同（相等），记录比较见 `RecordComparator`.
`static List<Record>`	`readRecords(File dir)` 从给定的本地目录读取记录.
`TaskOutput`	`runMapper(JobConf job, MapUTContext context)` 执行 `Mapper` 过程.
`TaskOutput`	`runReducer(JobConf job, ReduceUTContext context)` 执行 `Reducer` 过程.
`static void`	`writeRecords(File dir, List<Record> records, TableMeta meta)` 将记录写到给定的本地目录.

Methods inherited from class java.lang.Object
equals, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

- Constructor Detail
  - MRUnitTest
```
public MRUnitTest()
```
- Method Detail
  - runMapper
```
public TaskOutput runMapper(JobConf job,
                            MapUTContext context)
                     throws IOException,
                            ClassNotFoundException,
                            InterruptedException
```
    执行 Mapper 过程.
    本方法根据给定的 JobConf 和 MapUTContext 执行 Mapper 过程。
    
    Parameters:
    
    job - 作业描述
    
    context - 执行 Mapper 过程的上下文对象
    
    Returns:
    
    Mapper 输出
    
    Throws:
    
    IOException
    
    ClassNotFoundException
    
    InterruptedException
  - runReducer
```
public TaskOutput runReducer(JobConf job,
                             ReduceUTContext context)
                      throws IOException,
                             ClassNotFoundException,
                             InterruptedException
```
    执行 Reducer 过程.
    本方法根据给定的 JobConf 和 ReduceUTContext 执行 Reducer 过程。
    
    Parameters:
    
    job - 作业描述
    
    context - 执行 Reducer 过程的上下文对象
    
    Returns:
    
    Reducer 输出
    
    Throws:
    
    IOException
    
    ClassNotFoundException
    
    InterruptedException
  - createRecord
```
public static Record createRecord(String schema)
                           throws IOException
```
    给定 schema 创建记录对象（Record）.
    schema 的格式是：(<列名>:<类型>)(,<列名>:<类型>)+
    例如：a:string,b:string
    
    Parameters:
    
    schema - 待创建记录对象的 schema
    
    Returns:
    
    记录对象
    
    Throws:
    
    IOException
  - readRecords
```
public static List<Record> readRecords(File dir)
                                throws IOException
```
    从给定的本地目录读取记录.
    目录需要包含一个名为“__schema__”的文件描述表的 schema，格式为：[.,]:(,:)*
    例如：
    - proj.tablename,a:string,b:bigint,c:double
    - a:string,b:bigint,c:double
    Parameters:
    
    dir - 本地目录
    
    Returns:
    
    记录对象 List
    
    Throws:
    
    IOException
    
    See Also:
    
    #writeRecords(File, List, String)}
  - writeRecords
```
public static void writeRecords(File dir,
                                List<Record> records,
                                TableMeta meta)
                         throws IOException
```
    将记录写到给定的本地目录.
    需要给一个 schema，格式为：[.,]:(,:)*
    例如：
    - proj.tablename,a:string,b:bigint,c:double
    - a:string,b:bigint,c:double
    schema 会输出到给定目录下的一个名为“__schema__”的文件。
    Parameters:
    
    dir - 本地目录
    
    records - 待输出的记录对象 List
    
    schema - 记录的 schema
    
    Throws:
    
    IOException
    
    See Also:
    
    #readRecords(File)}
  - equalRecords
```
public static boolean equalRecords(List<Record> records1,
                                   List<Record> records2,
                                   boolean sort)
```
    比较两个记录列表是否相同（相等），记录比较见 RecordComparator.
    
    Parameters:
    
    records1 - 记录列表1
    
    records2 - 记录列表2
    
    sort - 是否先对列表排序后再比较
    
    Returns:
    
    记录列表相等返回true，否则返回false
  - equalRecords
```
public static boolean equalRecords(File dir,
                                   List<Record> records,
                                   boolean sort)
                            throws IOException
```
    比较给定本地目录下的记录集合和一个记录列表是否相同（相等），记录比较见 RecordComparator.
    此方法等价于使用 readRecords(File) 读取本地目录的记录对象到一个记录列表，然后再使用 equalRecords(List, List, boolean) 进行比较。
    
    Parameters:
    
    dir - 本地目录
    
    records - 记录列表
    
    sort - 是否先对列表排序后再比较
    
    Returns:
    
    Throws:
    
    IOException

Class MRUnitTest

Constructor Summary

Method Summary

Methods inherited from class java.lang.Object

Constructor Detail

MRUnitTest

Method Detail

runMapper

runReducer

createRecord

readRecords

writeRecords

equalRecords

equalRecords