ODPS开发大全：进阶篇

参数类型

具体使用

1. Map设置

set odps.sql.mapper.cpu=100

作用：设置处理Map Task每个Instance的CPU数目，默认为100，在[50,800]之间调整。场景：某些任务如果特别耗计算资源的话，可以适当调整Cpu数目。对于大多数Sql任务来说，一般不需要调整Cpu个数的。

set odps.sql.mapper.memory=1024

作用：设定Map Task每个Instance的Memory大小，单位M，默认1024M，在[256,12288]之间调整。场景：当Map阶段的Instance有Writer Dumps时，可以适当的增加内存大小，减少Dumps所花的时间。

set odps.sql.mapper.merge.limit.size=64

作用：设定控制文件被合并的最大阈值，单位M，默认64M，在[0,Integer.MAX_VALUE]之间调整。场景：当Map端每个Instance读入的数据量不均匀时，可以通过设置这个变量值进行小文件的合并，使得每个Instance的读入文件均匀。一般会和odps.sql.mapper.split.size这个参数结合使用。

set odps.sql.mapper.split.size=256

作用：设定一个Map的最大数据输入量，可以通过设置这个变量达到对Map端输入的控制，单位M，默认256M，在[1,Integer.MAX_VALUE]之间调整。场景：当每个Map Instance处理的数据量比较大，时间比较长，并且没有发生长尾时，可以适当调小这个参数。如果有发生长尾，则结合odps.sql.mapper.merge.limit.size这个参数设置每个Map的输入数量。

2. Join设置

set odps.sql.joiner.instances=-1

作用: 设定Join Task的Instance数量，默认为-1，在[0,2000]之间调整。不走HBO优化时,ODPS能够自动设定的最大值为1111，手动设定的最大值为2000，走HBO时可以超过2000。场景：每个Join Instance处理的数据量比较大，耗时较长，没有发生长尾，可以考虑增大使用这个参数。

set odps.sql.joiner.cpu=100

作用: 设定Join Task每个Instance的CPU数目，默认为100，在[50,800]之间调整。场景：某些任务如果特别耗计算资源的话，可以适当调整CPU数目。对于大多数SQL任务来说，一般不需要调整CPU。

set odps.sql.joiner.memory=1024

作用：设定Join Task每个Instance的Memory大小，单位为M，默认为1024M,在[256,12288]之间调整。场景：当Join阶段的Instance有Writer Dumps时，可以适当的增加内存大小，减少Dumps所花的时间。

3. Reduce设置

set odps.sql.reducer.instances=-1

作用: 设定Reduce Task的Instance数量，手动设置区间在[1,99999]之间调整。不走HBO优化时,ODPS能够自动设定的最大值为1111，手动设定的最大值为99999，走HBO优化时可以超过99999。场景：每个Join Instance处理的数据量比较大，耗时较长，没有发生长尾，可以考虑增大使用这个参数。

set odps.sql.reducer.cpu=100

作用：设定处理Reduce Task每个Instance的Cpu数目，默认为100，在[50,800]之间调整。场景：某些任务如果特别耗计算资源的话，可以适当调整Cpu数目。对于大多数Sql任务来说，一般不需要调整Cpu。

set odps.sql.reducer.memory=1024

作用：设定Reduce Task每个Instance的Memory大小，单位M，默认1024M，在[256,12288]之间调整。场景：当Reduce阶段的Instance有Writer Dumps时，可以适当的增加内存的大小，减少Dumps所花的时间。

上面这些参数虽然好用，但是也过于简单暴力，可能会对集群产生一定的压力。特别是在集群整体资源紧张的情况下，增加资源的方法可能得不到应有的效果，随着资源的增大，等待资源的时间变长的风险也随之增加，导致效果不好！因此请合理的使用资源参数！

4. 小文件合并参数

set odps.merge.cross.paths=true|false

作用：设置是否跨路径合并，对于表下面有多个分区的情况，合并过程会将多个分区生成独立的Merge Action进行合并，所以对于odps.merge.cross.paths设置为true，并不会改变路径个数，只是分别去合并每个路径下的小文件。

set odps.merge.smallfile.filesize.threshold = 64

作用：设置合并文件的小文件大小阀值，文件大小超过该阀值，则不进行合并，单位为M，可以不设，不设时，则使用全局变量odps_g_merge_filesize_threshold，该值默认为32M，设置时必须大于32M。

set odps.merge.maxmerged.filesize.threshold = 256

作用：设置合并输出文件量的大小，输出文件大于该阀值，则创建新的输出文件，单位为M，可以不设，不设时，则使用全局变odps_g_max_merged_filesize_threshold，该值默认为256M，设置时必须大于256M。

set odps.merge.max.filenumber.per.instance = 10000

作用：设置合并Fuxi Job的单个Instance允许合并的小文件个数，控制合并并行的Fuxi Instance数，可以不设，不设时，则使用全局变量odps_g_merge_files_per_instance，该值默认为100，在一个Merge任务中，需要的Fuxi Instance个数至少为该目录下面的总文件个数除以该限制。

set odps.merge.max.filenumber.per.job = 10000

作用：设置合并最大的小文件个数，小文件数量超过该限制，则超过限制部分的文件忽略，不进行合并，可以不设，不设时，则使用全局变量odps_g_max_merge_files，该值默认为10000。

5. UDF相关参数

set odps.sql.udf.jvm.memory=1024

作用: 设定UDF JVM Heap使用的最大内存，单位M，默认1024M，在[256，12288]之间调整。场景：某些UDF在内存计算、排序的数据量比较大时，会报内存溢出错误，这时候可以调大该参数，不过这个方法只能暂时缓解，还是需要从业务上去优化。

set odps.sql.udf.timeout=1800

作用：设置UDF超时时间，默认为1800秒，单位秒。[0，3600]之间调整。

set odps.sql.udf.python.memory=256

作用：设定UDF python 使用的最大内存，单位M，默认256M。[64，3072]之间调整。

set odps.sql.udf.optimize.reuse=true/false

作用：开启后，相同的UDF函数表达式，只计算一次，可以提高性能，默认为True。

set odps.sql.udf.strict.mode=false/true

作用：True为金融模式，False为淘宝模式，控制有些函数在遇到脏数据时是返回NULL还是抛异常，True是抛出异常，False是返回null。

6. Mapjoin设置

set odps.sql.mapjoin.memory.max=512

作用：设置Mapjoin时小表的最大内存，默认512,单位M，[128，2048]之间调整。

7. 动态分区设置

set odps.sql.reshuffle.dynamicpt=true/false

作用：默认true，用于避免拆分动态分区时产生过多小文件。如果生成的动态分区个数只会是很少几个,设为false避免数据倾斜。

8. 数据倾斜设置

set odps.sql.groupby.skewindata=true/false

作用：开启Group By优化。

set odps.sql.skewjoin=true/false

作用：开启Join优化，必须设置odps.sql.skewinfo 才有效。

函数类型

说明

日期函数

支持处理DATE、DATETIME、TIMESTAMP等日期类型数据，实现加减日期、计算日期差值、提取日期字段、获取当前时间、转换日期格式等业务处理能力。

数学函数

支持处理BIGINT、DOUBLE、DECIMAL、FLOAT等数值类型数据，实现转换进制、数学运算、四舍五入、获取随机数等业务处理能力。

窗口函数

支持在指定的开窗列中，实现求和、求最大最小值、求平均值、求中间值、数值排序、数值偏移、抽样等业务处理能力。

聚合函数

支持将多条输入记录聚合成一条输出值，实现求和、求平均值、求最大最小值、求平均值、参数聚合、字符串连接等业务处理能力。

字符串函数

支持处理STRING类型字符串，实现截取字符串、替换字符串、查找字符串、转换大小写、转换字符串格式等业务处理能力。

复杂类型函数

支持处理MAP、ARRAY、STRUCT及JSON类型数据，实现去重元素、聚合元素、元素排序、合并元素等业务处理能力。

加密函数

支持处理STRING、BINARY类型的表数据，实现加密、解密等业务处理能力。

其他函数

除上述函数之外，提供支持其他业务场景的函数。

函数名

具体操作

1. 获取当前日期7天内的日期：

SELECTDATEADD(GETDATE(),-7,'dd');

TO_CHAR(DATEADD(GETDATE(),-7,'dd'),'yyyymmdd');

2. DATEADD（指定日期加减）:

SELECT DATEADD(GETDATE(),1,"dd"); //2021-01-09 10:48:40 GETDATE()返回值是2021-01-08 10:48:40

SELECT DATEADD(GETDATE(),1,"mm");//2021-02-08 10:49:24 GETDATE()返回值是2021-01-08 10:49:24

3. DATEDIFF（计算两个日期的差值）：

datediff(end, start, 'dd') = 1

datediff(end, start, 'mm') = 1

datediff(end, start, 'yyyy') = 1

datediff(end, start, 'hh') = 1

4. DATEPART（返回指定日期的年/月/日）：

SELECT DATEPART(GETDATE(),'mm');

SELECT DATEPART(GETDATE(),'yyyy');

SELECT DATEPART(GETDATE(),'dd');

SELECT DATEPART(GETDATE(),'hh');

5. DATETRUNC（截取时间）：

datetrunc(datetime '2011-12-07 16:28:46', 'yyyy') = 2011-01-01 00:00:00

datetrunc(datetime '2011-12-07 16:28:46', 'month') = 2011-12-01 00:00:00

datetrunc(datetime '2011-12-07 16:28:46', 'DD') = 2011-12-07 00:00:00

6. TO_CHAR 函数

使用方式 to_char（要处理的日期,日期格式）

推荐用法:2018-01-11 10:00:00 格式 ，处理为指定格式的日期字符串

效果：处理为yyyymmdd的日期格式，类型为字符串

to_char('2018-01-11 10:00:00','yyyymmdd') as date_3 

to_char('2018-01-11 10:00:00','yyyymmdd hh:mi:ss') as date_5 

to_char('2018-01-11 10:00:00','yyyy-mm-dd hh:mi:ss') as date_6

to_char('2018-01-11 10:00:00','yyyy-mm-dd 00:00:00') as date_8 

to_char('2018-01-11 10:00:00','yyyy-mm-01 23:59:59') as date_9 

7. TO_DATE函数

使用方式:to_date(datetime,format)

推荐用法:根据时间的格式，适当调整format的模版

效果：处理20180111、2018-01-11、2018-01-11 10:00:00

to_char('2018-01-11 10:00:00','yyyymmdd') as date_3

to_char('2018-01-11 10:00:00','yyyymmdd hh:mi:ss') as date_5

to_char('2018-01-11 10:00:00','yyyy-mm-dd hh:mi:ss') as date_6

to_char('2018-01-11 10:00:00','yyyy-mm-dd 00:00:00') as date_8

8. UNIX时间戳转换

函数：datetime from_unixtime(bigint unixtime) 支持秒

from_unixtime(123456789) = 1973-11-30 05:33:09

函数: from_utc_timestamp(bigint unixtime,string timezone) 支持毫秒

SELECT from_utc_timestamp(1501557840000 ,'GMT') ;

--返回：2017-08-01 04:24:00

9. DATE转UNIX时间戳

函数：bigint unix_timestamp(datetime date)

select unix_timestamp(datetime '2019-09-20 01:00:00'); 

--返回1568912400

select unix_timestamp('2019-09-20 01:00:00'); 

--返回1568912400

阶段

特征

原因

解决方案

调度阶段

子状态为“Waiting for cluster resource”，作业排队等待被编译。

1.计算集群资源紧缺。

查看计算集群的状态，需要等待计算集群的资源。

2. 编译资源池资源不够

优化阶段

子状态为“SQLTask is optimizing query”，优化器正在优化执行计划。

1.执行计划复杂，需要等待较长时间做优化。

一般可接受10分钟以内，如果真的太长时间不退出，基本可认为是 odps 的 bug。

生成物理执行计划阶段

子状态为“SQLTask is generating execution plan”。

1.读取的分区太多。每个分区需要去根据分区信息来决定处理方式，决定 split，并且会写到生成的执行计划中。

需要好好设计 SQL，减少分区的数量，包括：分区裁剪、筛除不需要读的分区、把大作业拆成小作业。

2.小文件太多（万级别），ODPS 会根据文件大小决定 split，小文件多了会导致计算 split 的过程耗时增加。

使用TunnelBufferedWriter接口，可以更简单的进行上传功能，同时避免小文件。

执行一次 alter table merge smallfiles; 让 odps 把小文件 merge 起来，

数据跨集群复制阶段

子状态列表里面出现多次“Task rerun”，result 里有错误信息“FAILED: ODPS-0110141:Data version exception”。

1.project 刚做集群迁移，往往前一两天有大量需要跨集群复制的作业。

这种情况是预期中的跨集群复制，需要用户等待。

2.可能是作业提交错集群，或者是中间 project 做过迁移，分区过滤没做好，读取了一些比较老的分区。

检查作业提交的集群是否正确， Logview2.0任务详情页左侧的 BasicInfo 查看作业提交的集群。

阶段

特征

解决方案

等待资源

一些instance处于Ready状态，部分instance处于Running状态。

确定排队状态是否正常。可以通过 logview 的排队信息“Queue”看作业在队列的位置。

数据倾斜

task 中大多数 instance 都已经结束了，但有某几个 instance 却迟迟不结束（长尾）。

1. 利用 MaxCompute Studio 的作业执行图及作业详情功能来分析作业运行情况，定位到长尾实例，找到导致长尾的数据来源。

2. 利用 Logveiw2.0 查看任务执行图和 instance 运行情况来定位长尾实例。

UDF执行低效

某个 task 执行效率低，且该 task 中有用户自定义的扩展。

1. 检查 UDF 是否有 bug。

有时候 bug 是由于某些特定的数据值引起的，比如出现某个值的时候会引起死循环。

2. 检查 UDF 函数是否与内置函数同名。

内置函数是有可能被同名 UDF 覆盖的，当看到一个函数像是内置函数时，需要确定是否有同名 UDF 覆盖了内置函数。

3. 使用内置函数代替 UDF。

evaluate 中只做与参数相关的必要操作。

数据膨胀

task 的输出数据量比输入数据量大很多。

1. 检查代码是否有 bug：JOIN 条件是不是写错，变成笛卡尔积了；UDTF是不是有问题，输出太多数据。

2. 检查 Aggregation 引起的数据膨胀。

3. 避免join引起的数据膨胀。

4. 由于grouping set 导致的数据膨胀。

在线业务压制

ODPS集群中的一部分是离线集群，另一部分是在线集群。

如果是弹内环境，可通过fuxi sensor确认是否存在在线业务压制。

子查询多阶段执行

MaxCompute SQL 的子查询会被编译进同一个 Fuxi DAG，即所有子查询和主查询都通过一个 Fuxi 作业完成。

但也有一些特殊子查询需要先将子查询单独执行。

子查询 SELECT DISTINCT ds FROM t_ds_set 先执行，其结果需要被用来做分区裁剪，来优化主查询需要读取的分区数。

过多小文件

存储方面：小文件过多会给 Pangu 文件系统带来一定的压力，且影响空间的有效利用。

计算方面：ODPS 处理单个大文件比处理多个小文件更有效率，小文件过多会影响整体的计算执行性能。

为了避免系统产生过多小文件，SQL作业会在结束时自动触发合并小文件的操作。

根据参数odps.merge.smallfile.filesize.threshold来判定小文件，默认阈值为32MB。

可通过logview查看作业是否触发了自动合并小文件。

动态分区元数据更新

Fuxi 作业执行完后，有可能还有一些元数据操作。

对分区表 sales 使用 insert into ... values命令新增 2000 个分区：

INSERT INTO TABLE sales partition (ds)(ds, product, price)

VALUES ('20170101','a',1),('20170102','b',2),('20170103','c',3), ...;

输出文件size变大

在输入输出条数相差不大的情况，结果膨胀几倍。

一般是数据分布变化导致的，在写表的过程中，会对数据进行压缩，而压缩算法对于重复数据的压缩率是最高的。

优化类型

具体类型

优化措施

调整并行度

instance数量的增加会对执行速度产生影响：

1. 更多的instance意味着更长的等待资源和排队次数。

2. 每个instance的初始化需要一定时间，并行度越高，总初始化时间越长，有效执行时间占比越低。

需要强制 1 个 instance 执行

用户需要检查这些操作是否必要，能否去掉，尽量取消掉这些操作：读表的 task + 非读表的 task

影响单个task并行度主要因素：

1. 某些操作强制必须 1 个 instance 来执行

2. 读表的 task

3. 非读表的 task

4. HBO会在上面的基础上根据历史作业的执行情况做调整

对于读表的 task，一个 instance 读取 256M的数据，一些常见出问题的情况：

1. 数据压缩比很高

2. Task 中执行了一些很 heavy 的操作，特别是存在 UDF

3. 读取 256M 数据太少，导致 instance 的执行时间太短

可以通过调整flag实现：

set odps.sql.mapper.split.size= xxx

非读表的 task，主要有三种方式调整并行度：

1. 调整 odps.sql.mapper.split.size

2. 通过 odps.sql.reducer.instances 强制设置 reducer 并行度

3. 通过 odps.sql.joiner.instances 强制设置 joiner 并行度

set odps.sql.reducer.instances= xxx

-- 设定Reduce task的instance数量

set odps.sql.joiner.instances= xxx

-- 设定Join task的instance数量

HBO

HBO (History-Based Optimization) 会根据对历史作业的分析来优化当前作业的。

包括内存、并行度等一系列参数，它能让你的周期作业越跑越快。

为了尽可能解决HBO失效这个顽疾，我们在HBO中增加了若干新的功能，包括：

• realtime hbo

• task-wise hbo

• new signature

优化执行计划

CBO优化器会基于统计信息、SQL语义、执行引擎能力、丰富的优化能力，自动生成最优的执行计划，并且在持续提升优化能力。

Map Join Hint

用户可以手动添加map join hint，使得原本的Sort-Merge Join变成Map Join，避免大表数据shuffle从而提升性能。

Distributed Map Join Hint

Distributed MapJoin是MapJoin的升级版，适用于适用于大表Join中表的场景 的场景，二者的核心目的都是为了减少大表侧的Shuffle和排序。

Dynamic Filter Hint

基于JOIN等值连接的特性，MaxCompute可以通过表A的数据生成一个过滤器，在Shuffle或JOIN之前提前过滤表B的数据。

物化视图

物化视图(Materialized View)本质是一种预计算，即把某些耗时的操作（如JOIN/AGGREGATE）的结果保存下来。

以便在查询时直接复用，从而避免这些耗时的操作，最终达到加速查询的目的。

数据倾斜

数据Shuffle导致的数据倾斜 1

数据倾斜大多数是由于数据的 reshuffle 引起的，因为按照某个 key 来做 shuffle，同一个 key 值的数据会强制集中在一个 instance 处理。

1. 去掉 shuffle

2. 换别的 shuffle key

3. 将热点数据特殊处理

数据Shuffle导致的数据倾斜 2

特征：读表并写动态分区作业，M task 读入大量数据，但是只会写出少量的 动态分区。

解决方法：set odps.sql.reshuffle.dynamicpt =false; 去掉reshuffle 过程。

优化类型

具体类型

优化措施

SQL的新语法、新功能

GROUPING SETS：对 SELECT 语句中 GROUP BY 子句的扩展。

SQL 运行*时物理执行计划做了 3 次聚合，然后再 UNION 起来。

脚本模式：

脚本模式能让用户以脚本的形式提交多条语句同时执行。

脚本模式的性能优势，实际上是“将分散的业务逻辑合并成一个作业来运行“的性能优势：

1. 合并重复的公共操作。

2. 避免中间数据写表，减少临时表。

3. 更好的发挥 optimizer 的作用。

4. 减少了作业调度的开销。

MR典型场景用SQL实现

1. 使用 SQL-聚合函数

select k, WM_CONCAT(';',concat(v,":",c)) from 

( select k, v, count(v) c from t group by k,v) t2 group by k;

2. 用 SQL-窗口/分析函数

rows between x preceding|following and y preceding|following

3. 使用 SQL-UDJ (User Defined Join)

MapReduce 实现的 JOIN 逻辑。

4. 使用 SQL-TRANSFORM

适用场景：MapReduce Streaming 作业。

合理设置资源参数

Map设置

set odps.sql.mapper.cpu=100

作用：设置处理Map Task每个Instance的CPU数目

set odps.sql.mapper.memory=1024

作用：设定Map Task每个Instance的Memory大小

set odps.sql.mapper.merge.limit.size=64

作用：设定控制文件被合并的最大阈值

set odps.sql.mapper.split.size=256

作用：设定一个Map的最大数据输入量

Join设置

set odps.sql.joiner.instances=-1

作用: 设定Join Task的Instance数量

set odps.sql.joiner.cpu=100

作用: 设定Join Task每个Instance的CPU数目

set odps.sql.joiner.memory=1024

作用：设定Join Task每个Instance的Memory大小

Reduce设置

set odps.sql.reducer.instances=-1

作用: 设定Reduce Task的Instance数量

set odps.sql.reducer.cpu=100

作用：设定处理Reduce Task每个Instance的Cpu数目

set odps.sql.reducer.memory=1024

作用：设定Reduce Task每个Instance的Memory大小