使用阿里云ODPS之点滴

    趁着这次做网聚宝搜索服务化这个项目，体验了一下阿里云的重量级云产品ODPS，也就是传说中的云梯2。项目几近结束，现在回过头来看当初选择使用OPDS的决定还是相当正确的，ODPS在网聚宝搜索上云这个项目中起到了非常至关重要的作用，一句话总结：ODPS是非常靠谱的。从刚开始里了解ODPS开始，到引入项目中使用，进而通过优化加速在项目中使用得游刃有余，经历了一个过程，中间也走了不少弯路。在此将之前碰到问题点做一下总结，希望对有和我们一样需求的团队或者个人有所帮助。

 

终搜为什么选择ODPS

    搜索引擎都是为了解决业务方数据多维度随意组合的条件搜索的需求，终搜的产品也不例外，在构建索引的过程中有非常重要的一个步骤就是将业务数据库中的数据导入搜索引擎中。搜索引擎中保存的数据结构是单维度的记录，而网聚宝搜索上云项目中所用到的数据源是来自多个数据库的多张数据库表。对象实体是由一对多，多对多的表组成的。所以在数据导入到搜索引擎之前对数据进行扁平化处理成为一个必不可少的环节。

 

    实现这个操作在java社区有非常成熟的hadoop可以使用，基于hadoop之上的hive通过执行命令SQL化可以很好地屏蔽底层技术细节。这个是一个不错的选择，同时我们在方案选型的过程中了解到阿里集团在公有云中正在推广ODPS，并且已经开始在公有云中售卖。抱着试试看的想法，开启了ODPS之旅。

 

   其实现在总结起来，使用ODPS还有一个非常重要的原因，就是阿里云在云基础设施配套这块做得比较到位，比如，针对ODPS数据导入有CDP（Cloud Data pipeling）这个产品来负责数据导入工作。通过CDP，只需要简地将单数据的reader和writer通过XML描述发送给CDP中间件，就可以轻松将几个G甚至几个T的数据从RDS在短短几十分钟之内DUMP到ODPS中。有了CDP可以免去我们很多额外工作。

 

   在ODPS上执行多表数据扁平化工作，通俗地讲就是“打宽表”，举个最简单的例子，例如有一张会员表，一张交易表，两表的关系是一对多，那我们可以按照会员维度将两张表打成一张宽表。最终打好的宽表结构就变成，user.id,user.name,”trade.id,trade.id  …….” As tradeids。这对于ODPS来说小菜一碟，有现成的聚合函数可以使用，如wm_concat。终搜将宽表构建成索引，用户可以很方便地通过trade.id来反查user.id。这是一个最简单的例子，或许你会问处理这种问题数据库也是可以的。那我要提醒你，数据库处理的数量级和ODPS处理的数据级完全不在一个等级上。但凡数据上G上T之后，数据库只能干瞪眼了，而像ODPS这样的MR工具却可以轻松搞定，再多数据也只需要通过集群节点的水平扩容轻松搞定。

 

ODPS的收费

   对于收费这个问题，我们是使用的是集团弹外环境，按计算量来收费的，终搜因为是集团内部团队，可以通过申请ODPS抵价券来使用。基本上，打一个2亿条记录的宽表（需要关联多张表，且有比较复杂的聚合），一次收费大概是100元左右，在开发测试阶段是非常昂贵的，因为测试阶段往往是修改一点测试一下，确认结果是否符合预期。一个功能点往往要来来回回折腾好几次，因此而产生的费用可想而知了，所以在测试阶段不要用全部数据来进行测试开发，抽取一个样本子集来进行测试。这样，因为数据量少了，可以缩短计算时间，缩短测试周期，并且降低费用。

 

单个Select进行多维度统计

 

   在ODPS可以基于SQL来执行的，值得一提的是虽是SQL，但和一般关系数据库上执行的SQL还是有很多区别的。举一个例子，可以通过一个SQL语句来多维度统计。举个例子：有两张表，班级表和学生表，表结构：

1.    class（cid,name, dues(班会费),grade(年级)）
2.   student(sid,cid(班级外键),gender(性别))

 

   统计结果的结构如下：

班级名称

班级所在年级班会费总和

年级拥有的班级数据

班级男生人数

班级女生人数

 

   在这个统计中需要用两个维度来统计，年级，和班级，按照Mysql这样的关系型数据库是不可能用一个SQL语句得到统计结果，需要分几个SQL按照不同维度进行统计，最后将统计结果组装起来。但是在ODPS中可以利用ODPS的窗口函数或者聚合函数来统计，SQL如下：

 

SELECT c.name as class_name,

      SUM(c.dues ) OVER ( PARTITION BY c.grade )  dues_sum_grade,

      Count(c.id) OVER(PARTITION BY c.grade)   class_count _grade,

      Count( CASE WHEN s.gender=’f’ THEN 1 END ) as female_count,

      Count( CASE WHEN s.gender=’m’ THEN 1 END ) as male_count

FROM class c INNER JOIN student s ON c.cid = s.cid

GROUP BY c.cid, c.name

 

ODPS SQL 窗口函数（http://odps.alibaba-inc.com/doc/prddoc/odps_sql/odps_sql_func.html#id5 ）

ODPS SQL聚合函数（http://odps.alibaba-inc.com/doc/prddoc/odps_sql/odps_sql_func.html#odps-sql-aggr ）

大任务分解

    习惯于写在关系数据库上写SQL来得到统计结果的开发开发者，往往习惯于把一个统计过程写成一个SQL语句，但是随着统计逻辑变得复杂，这个SQL也会变得非常复杂，可读性也急剧下降。另外，调试也变得麻烦，在开发过程中的一个体验是，写SQL脚本稍有不慎，执行过程就会执行很长时间也不能得到结果，或者直接报错，往往从出错信息上很难看出到底哪儿写错了，让人一头雾水，往往只能凭感觉去纠错。

 

    解决的办法，是将一个大的SQL语句拆解成一个主表和多个分表，且主表和每张分表之间的关系要做到一对一，这里一对一的关系非常重要，因为只有一对一的关系才能使得各个表在join过程中出错的概率降到最低，在最后一步将主表和分表组合起来。该过程可以类比成汽车组装工厂的操作流程，各个车间先组装好每个模块构建，最后再由组装车间将每个构建组装起来。

 

    这样做有一个好处显而易见，那就是，非常容易定位问题，因为无论发生什么错误，都能将问题范围缩小到某个子语句中，这样能大大提高开发和维护的效率。

 

    将一个大的任务分而治之，分步骤执行，好处显而易见，同时也带来了一个问题需要将所有的各个子任务分步骤有序执行，因为有些子任务没有前后依赖关系可以并发执行，而有些任务有前后依赖关系，一个执行完了，才能触发另外一个执行。

 

防止聚合爆炸

       ODPS官方文档上提供了非常丰富的聚合函数，例如wm_concat,sum,count 等等。在使用聚合函数时，要注意一点，要防止聚合函数执行结果爆炸。举个例子，会员（user[userid,username]）和购买记录（trade[tradeid,userid,payment,item_title]）是一对多的关系。对这两张表以会员的维度进行聚合组合的记录为user_trade[userid,username,payment_sum,item_titles],item_titles 这个字段是用wm_concat(distinct trade.item_title)以userid分组聚合而成的，SQL如下：

SELECT userid,username,wm_concat(DISTINCT ‘,’,item_title) as item_titles

FROM user u

INNER JOIN trade t ON (u.userid = t.userid)

GROUP BY u.userid,u.username

 

    正常情况下，一个用户不会超过几十笔购买记录。但是，不得不承认林子大了什么鸟都会有，就有这样一些奇葩用户，有上万笔购买记录。导致ODPS脚本执行得非常慢，而且最终统计结果超过ODPS限定的字符串长度上限而使整个计算过程失败。

解决办法是需要将这些奇葩用户在计算之前先过滤掉，在计算之前先需要生成一张中间表用来表示所有合法的用户id集合，中间表SQL如下：

CREATE TABLE tmp_user_filter AS

SELECT tt.userid

 FROM(

SELECT userid ,count(t.tradeid) as tcount

FROM user u INNER JOIN user_trade t ON(u.userid = t.userid)

GROUP BY u.userid )tt

WHERE tt.tcount < 200

     通过这个SQL创建了一个临时表，将所有所有交易笔数少于200笔交易的用户过筛出来，当然这个200这个阀值可以按照业务的需要进行调节。之后在其他脚本中可以与tmp_user_filter这个表进行INNER JOIN，再用聚合函数统计，那就保证不会聚合爆炸了。

改造后的SQL如下：

SELECT userid,username,wm_concat(DISTINCT ‘,’,item_title) as item_titles

FROM user u

   INNER JOIN tmp_user_filter f ON (u.userid = f.userid)

INNER JOIN trade t ON (u.userid = t.userid)

GROUP BY u.userid,u.username

 

巧妙使用row_number()函数

     row_number()函数是一个不错的工具，能解决很多场景下的实际问题。举个例子在mysql中，有一张表student(sid,name,gender ,score)，要求班级中男生，女生考试分数最高同学的姓名及分数。这个SQL用mysql来写也很费劲必须要依赖表中的一个外键，需要依赖外键的唯一性先求出id，然后再进行一次JOIN查询（http://stackoverflow.com/questions/12699799/mysql-selecting-max-record-in-group-by

）

   在ODPS中这个问题可以使用row_number()函数轻松搞定，sql如下：

SELECT aa.sid,aa.name,aa.gender,aa.score

 FROM(

SELECT row_number() over(partition by gender order by score desc) AS row_num,

       sid,name,gender,score

FROM student) aa

WHERE aa.row_num<2

    举一反三，也可以查询分数排前三的同学的记录，只需要改变WHERE条件的row_num值即可。

 

合理使用partition加速计算

       ODPS是基于文件系统的，存在ODPS中的数据会以partition(分区)为单位存在ODPS的计算集群的节点中。在ODSP执行计算逻辑时，能够有效提升计算速度的原则是，尽可能将数据分成足够多的分区，将单个分区的数据量足够小。写SQL的时候避免跨分区做join计算。

比如有a，b两张表，他们的分区策略是相同，a、b表之间存在一对多的逻辑关系。表的分区键是‘ps’。

Select *

From a inner join b on (a.id= b.f_id ANDa.ps=b.ps)

WHERE a.ps=’xxx’

    这样写虽然最后执行结果是正确的，但是需要花比较长的时间，原因是对于b表扫描的目标记录不止‘xxx’这一个分区而是b表的全部分区，正确的写法应该是，如下：

 

Select *

From a inner join b on (a.id= b.f_id )

WHERE a.ps=’xxx’ and b.ps=’xxx’