public class MRUnitTest extends Object
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 and Description |
|---|
MRUnitTest() |
| 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)
将记录写到给定的本地目录.
|
public TaskOutput runMapper(JobConf job, MapUTContext context) throws IOException, ClassNotFoundException, InterruptedException
Mapper 过程.
本方法根据给定的 JobConf 和 MapUTContext 执行 Mapper 过程。
job - 作业描述context - 执行 Mapper 过程的上下文对象Mapper 输出IOExceptionClassNotFoundExceptionInterruptedExceptionpublic TaskOutput runReducer(JobConf job, ReduceUTContext context) throws IOException, ClassNotFoundException, InterruptedException
Reducer 过程.
本方法根据给定的 JobConf 和 ReduceUTContext 执行 Reducer 过程。
job - 作业描述context - 执行 Reducer 过程的上下文对象Reducer 输出IOExceptionClassNotFoundExceptionInterruptedExceptionpublic static Record createRecord(String schema) throws IOException
Record).
schema 的格式是:(<列名>:<类型>)(,<列名>:<类型>)+
例如:a:string,b:string
schema - 待创建记录对象的 schemaIOExceptionpublic static List<Record> readRecords(File dir) throws IOException
目录需要包含一个名为“__schema__”的文件描述表的
schema,格式为:[
例如:
dir - 本地目录ListIOException#writeRecords(File, List, String)}public static void writeRecords(File dir, List<Record> records, TableMeta meta) throws IOException
需要给一个 schema,格式为:[
例如:
schema 会输出到给定目录下的一个名为“__schema__”的文件。
dir - 本地目录records - 待输出的记录对象 Listschema - 记录的 schemaIOException#readRecords(File)}public static boolean equalRecords(List<Record> records1, List<Record> records2, boolean sort)
RecordComparator.records1 - 记录列表1records2 - 记录列表2sort - 是否先对列表排序后再比较public static boolean equalRecords(File dir, List<Record> records, boolean sort) throws IOException
RecordComparator.
此方法等价于使用 readRecords(File) 读取本地目录的记录对象到一个记录列表,然后再使用
equalRecords(List, List, boolean) 进行比较。
dir - 本地目录records - 记录列表sort - 是否先对列表排序后再比较IOExceptionCopyright © 2020 Alibaba Cloud Computing. All rights reserved.