MaxCompute（原ODPS） MapReduce常见问题解答

本文用到的

阿里云数加－大数据计算服务MaxCompute产品地址：https://www.aliyun.com/product/odps

1. 作业出现ClassNotFoundException和NoClassDefFoundError异常失败？

A: 对于ClassNotFoundException异常，一般是依赖的class不在你的jar包中，需要把依赖的库打到作业jar包中或者单独上传，并在-resources中指定；
对于NoClassDefFoundError异常，先看看依赖class是否存在于你的jar包，很多情况下是由于版本冲突导致的，可能你依赖的库和服务端自带的jar有冲突。

2. MR提交命令中-resources和-classpath的理解？

A: 在MaxCompute中类似MR这类分布式数据处理框架，用户的代码一般在以下两个地点执行：

• 运行客户端的进程／子进程：这里的用户代码负责准备执行环境、配置任务参数、提交任务，入口通常是main class。它不受沙箱限制，执行逻辑由用户代码驱动。同样的，这里的classpath由用户配置，或在console中使用-classpath选项添加依赖的classpath路径。
• 远程执行的worker进程：这里的代码负责执行数据处理逻辑，入口是mapper／reducer class。它受限沙箱限制，且执行逻辑由MaxCompute框架驱动。用户在命令行配置的-classpath在这里无效（显然，远程机器的路径和客户端机器的路径无法保证一致），任何第三方依赖必须作为resource提前上传至MaxCompute，并在提交任务时使用-resources选项或JobConf.setResources(String[])来设定。

3. Mapper数目如何设置？

A：如果没有输入表是可以直接指定map数目setNumMapTasks
    有输入表的话，setNumMapTasks不生效，需要通过setSplitSize来控制map数，默认是256M。

4. Reducer数目如何设置？

A: 通过JobConf的接口setNumReduceTasks可以设置。
对于pipeline作业，Jobconf的接口同样可以设置，只不过设置后所有reduce阶段的个数都是同样的值。
如果要分阶段设置，设置方式如下：

    Pipeline pipeline = Pipeline.builder()

.addMapper(TokenizerMapper.class)
...
.addReducer(SumReducer.class).setNumTasks(5)
...
.addReducer(IdentityReducer.class).setNumTasks(1).createPipeline();

5. 报错java.lang.OutOfMemoryError: Java heap space，MR的内存设置问题？

A：mapper或reducer的内存由两部分组成，JVM的heap memory和JVM 之外的框架相关内存。
    设置JVM内存的接口是（都是Java逻辑的话，调节内存是用下面两个接口）：
    setMemoryForMapperJVM和setMemoryForReducerJVM (默认是1024 单位MB)
    设置框架内存（c++部分的）的接口是（一般不需要设置）：
    setMemoryForMapTask和setMemoryForReduceTask（默认是2048 单位MB）

6. mr 输出到表或某个分区里时，输出的模式时追加还是覆盖 ？

A： 会覆盖输出表或分区之前的内容

7. 二次排序功能，MR相关配置解释，setMapOutputKeySchema? setOutputKeySortColumns? setPartitionColumns? setOutputGroupingColumns?

A: 通常情况下，GroupingColumns包含在KeySortColumns中，KeySortColumns和PartitionColumns要包含在Key schema中。

• 在Map端，Mapper输出的Record会根据设置的PartitionColumns计算哈希值，决定分配到哪个Reducer，会根据KeySortColumns对Record进行排序。
• 在Reduce端，输入Records在按照KeySortColumns排序好后，会根据GroupingColumns指定的列对输入的Records进行分组，即会顺序遍历输入的Records，把GroupingColumns所指定列相同的Records作为一次reduce函数调用的输入。

8. 请问mr job的map或者reduce如果想提前终止job, 执行什么代码？

A: 抛异常就可以，例如throw new RuntimeException("XXX"); 会导致job失败，job也就结束了。

9. 请问map阶段有时候为什么会有interrupted，但是map 最终仍然完成了？

A：因为有backup instance在跑，产生backup instance一般是因为有某些map instances明显慢于其他的，就会在别的机器上启动一个同样的worker来跑，这个功能类似于hadoop的预测执行，只要其中某个成功跑完，其他的就可以停掉了(变为interrupted)

10. mr如何获取输入表的信息？

A： 参考：http://repo.aliyun.com/java-sdk-doc/com/aliyun/odps/mapred/Mapper.TaskContext.html
使用Mapper.TaskContext的接口getInputTableInfo()，会得到输入表的TableInfo对象
每个map worker只会处理来自单一表或分区的数据，在mapper的setup阶段获取该信息即可。

11. 如何使用自定义partitioner ？

A: 参考如下代码：

import com.aliyun.odps.mapred.Partitioner;

...

public static class MyPartitioner extends Partitioner {

@Override
public int getPartition(Record key, Record value, int numPartitions) {
  // numPartitions即对应reducer的个数
  // 通过该函数决定map输出的key value去往哪个reducer
  String k = key.get(0).toString();
  return k.length() % numPartitions;
}
}

在jobconf里进行设置：jobconf.setPartitionerClass(MyPartitioner.class)
另外需要在jobconf里明确指定reducer的个数：jobconf.setNumReduceTasks(num)

12. 如何设置Key排序列的顺序（ASC or DESC）?

A: 类似如下： 
//key按这些列排序
job.setOutputKeySortColumns(new String[] { "custid", "msgtype","amount" });
//设置每个列正序还是倒序
job.setOutputKeySortOrder(new SortOrder[]{SortOrder.ASC,SortOrder.ASC,SortOrder.DESC});

13. 报错kInstanceMonitorTimeout, usually caused by bad udf performance，怎么解决？

A: 报这个错的原因是mapper或者reducer有逻辑执行时间特别长，且没有从输入表的读数据或者写出数据，超过默认10min后，会报这个异常；有两种解决方法：

• 将超时的时间调的更长一些，可以设置参数odps.function.timeout或者设置JobConf#setFunctionTimeout，最长可以设置为3600，即一个小时。
• 定期向框架汇报心跳 TaskContext#progress()，注意progress不要调用过于频繁，否则有性能问题，能确保两次调用之间的时间小于设置的timeout时间即可。

14. 框架map或者reduce接口里的Record对象是复用的？

A：是的，为了减少对象的开销，框架对于map, reduce接口里的Record对象是复用的，也就是说每次map或者reduce的每次迭代，Record对象没有变，只是里面的数据变化了。如果要保存上一次的Record需要toArray()拿到里面的数据对象进行保存。具体可以参考：http://repo.aliyun.com/java-sdk-doc/com/aliyun/odps/mapred/Reducer.html

15. 写完一条记录后，想把outputRecord里面的数据清空，这个要怎么弄，要不然，再写下一条记录的时候，如果某个字段没有值，就会用原来的记录填充?

    A：如果写的Record对象是复用的，如果某个值没有新set，则还是保留着之前的值对象。目前没有直接可以清空的api可以用，可以通过Record.getColumnCount拿到column count，用一个for 循环去一一set null即可。

16. MR支持多路输入输出，应该怎么写这样的程序？

    A：参考：多路输入输出示例

对于多路输入，每个输入源对应单独的一个Map阶段，即一个map task只会读取一个输入表的数据。可以指定一个表的多级分区列来作为一个输入，例如a, b, c三分区列，指定分区时可以指定a=1/b=1/c=2类似这样。

    如果同一级别的多个分区，则需要各自作为单独的分区输入，例如一个表的a=1和a=3分区作为多路输入的俩不同的输入，需要分别指定。
    maponly的作业也同样支持多路输入输出，实现方法类似。

17. sdk如何通过instance获取logview url？

A: 可以使用如下的方式拿到logview的url

RunningJob rj = JobClient.runJob(job);
com.aliyun.odps.Instance instance = SessionState.get().getOdps().instances().get(rj.getInstanceID());
String logview = SessionState.get().getOdps().logview().generateLogView(instance, 7 * 24);
System.out.println(logview);

18.  MR作业如何指定输入表的Project名字？

A: 可以按如下的方式指定：

InputUtils.addTable(TableInfo.builder().projectName("test_project_name").tableName("test_table_name").build(), job);

通过TableInfo.builder()的projectName接口来指定，如果不指定，默认值是在运行MR作业的那个project.

19. 不同的Mapper或者Reducer如何获取可区分的ID？

A: 有些业务场景需要区分不同的Mapper或Reducer，可以通过TaskContext的getTaskID接口获取到一个Mapper/Reducer独有的id。

String id = context.getTaskID().toString();

20. MR代码里有JNI的调用该怎么写？

A：首先project要开通jni的相关权限，在编译准备好so文件后，需要将so以file类型的形式添加为Resource，并在MR作业提交的时候-resources参数里指定，例如：

add file libtestjni.so as libtestjni.so -f;
jar -resources testmr.jar,libtestjni.so -classpath testmr.jar Test.MRDriver xxx xxx;

在MR的java代码使用jni的时候要注意，使用方式如下：

System.loadLibrary("testjni");    // 这里不要写成libtestjni.so,否则会报错，原因是java会自动添加lib前缀和.so后缀的

jni的使用方法可以参考：https://www.ibm.com/developerworks/cn/java/l-linux-jni/

21. MR作业读取表资源，Archive资源应该如何操作？

A: MaxCompute上的资源（file, table, archive等）可以类比于Hadoop的DistributedCache来理解，同样是会分发到每个计算节点上去，worker再从本地来读取，因而资源文件不能过大，否则分发资源就是一个瓶颈，目前默认有2G的总资源大小限制。
读取资源表，Archive资源总体上来说和读取file类型资源是类似的，只是使用的接口不同。读取资源文件的方法可以参考文档：使用资源示例

对于表资源：
将表添加为资源表： add table xxx as xxx -f;
读资源表的接口为：TaskContext#readResourceTable

对于Archive资源：
将本地archive(.tar, .zip等archive文件)上传为资源： add archive as xxx -f;
读archive资源的接口为：TaskContext#readResourceArchiveAsStream

...不断更新中...