本文用到的
阿里云数加-大数据计算服务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
…不断更新中…