从一个函数的三次迭代得到的收获

Photo by Frans Van Heerden from Pexels

学习一种新的编程语言，总得经历从陌生到熟悉，再到熟练，再到精通的过程。随着代码写得越来越多，有那么一段时间感觉上会认为自己进步了，但感觉这事往往不太靠谱。

那有没有什么量化的指标来显示自己的进步呢？

目前我想到两种：

1. 开发时间。面对同样工作量的开发任务，若是所花的时间较之前短了，那说明进步了。
2. 函数的迭代次数。一个合格的项目代码至少是由多个函数组成，而不仅仅只有一个主函数。每个函数都表示一个特定的功能，函数的迭代表示使用更精简、更易懂的代码完成同样的功能。
   每次迭代，都能表示你更熟悉当前的语言，掌握了更多的技巧。

在前期，应该更注重第二种方式。不是说开发时长不重要，而是在前期，作为新手，开发时长一定会快速的下降。甚至不需要多少主动学习，只要多写几行代码，很自然的就能把开发时间降下来。

而对一个函数进行迭代更需要主动性。迭代前的函数也能完成任务，为什么还要花时间去优化呢？你越主动学习新的知识，越花时间思考怎么应用新学的知识到代码里，那你的能力就提升的越快。

下面介绍下我对一个函数的三次迭代以及收获。

最近几周都在做类目相关的推荐，算法中需要写个函数用来获得每个类目id下gmv最高的商品。

在接手这个项目时，原开发者是用spark-sql写的代码：

// 获取各商品的GMV值
val goodsGMVRaw = spark.sql( 
  "select goods_id, sum(goods_revenue) as gmv " +
    "from table_name group by goods_id "
)

// 商品GMV信息中添加类目信息
val goodsGMV = goodsGMVRaw
  .map(line => (line(0).toString, brGoods2Catid.value(line(0).toString), line(1).toString.toDouble))
  .toDF("goods_id", "cat_id", "gmv")
goodsGMV.createTempView("goods_gmv")
println("goods_gmv finish")

// 每个类目根据gmv排序
val goodsGMVBest = spark.sql( 
  "select t.goods_id, t.cat_id from " +
    "(select goods_id, cat_id, row_number() over(partition by cat_id order by gmv desc) as rank from goods_gmv)t " +
    "where t.rank = 1 "
)

第一次迭代
由于 spark sql 运算会占用更多的内存，接手这个项目后我先把算法中关于 sql 的操作替换成 RDD 操作。除了占用资源会更少，后续的算法优化也会更方便些。

当时相法很简单，sql 中是通过 row_number 来获得 gmv 最高的商品的，那我通过 sortBy 来获得和row_number 同样的效果就行了。于是我修改成下面这样：

/**
  * 获取每个catid下被gmv最高的一个商品
  * @param spark
  * @param order_goods_info （goodsid, goods_revenue）
  * @param brGoods2Catid goodsid -> cat_id
  * @return (catid, goodsid)
  */
def getGoodsGmvTop(spark: SparkSession,
    order_goods_info: RDD[(String, String)], 
    brGoods2Catid: Broadcast[Map[String, String]]): RDD[(String, String)] ={
  order_goods_info.map(x => (x._1,x._2.toDouble))
    .reduceByKey(_+_)
    .map(x => (brGoods2Catid.value(x._1),x._1+","+x._2.toString))
    .reduceByKey(_+":"+_)
    .mapValues(x => {
      x.split(":")
        .map(p => (p.split(",")(0), p.split(",")(1).toFloat))
        .toList
        .sortBy(_._2)
        .take(1)
        .map(p => p._1)
        ._toString()
    })
}

该函数的想法是通过第一个reduceByKey获得各个商品的gmv信息；填加类目信息后通过第二个reduceByKey来获得一个商品列表，对其排序后取最gmv最大的商品。

第二次迭代
人的思维很容易被当前看到的东西所限制，第一次迭代时我用了sortBy函数，为的是实现row_number类似的功能。

但事实上我只需要获得gmv最大的商品就行，而不需要对整个商品列表进行排序。于是我就用maxBy函数代替了sortBy，使代码更加高效：

def getGoodsGmvTop(spark: SparkSession,
    order_goods_info: RDD[(String, String)], 
    brGoods2Catid: Broadcast[Map[String, String]]): RDD[(String, String)] ={
  order_goods_info.map(x => (x._1,x._2.toDouble))
    .reduceByKey(_+_)
    .map(x => (brGoods2Catid.value(x._1),x._1+","+x._2.toString))
    .reduceByKey(_+":"+_)
    .mapValues(x => {
      x.split(":")
        .map(p => (p.split(",")(0), p.split(",")(1).toFloat))
        .toList
        .maxBy(_._2)
        ._1
    })
}

第三次迭代
虽然上一次迭代用了maxBy函数，但我的思维还是被排序所限制，或者是因为对reduceByKey的使用还不熟。

我只需要保留每个类目Gmv最高的商品，所以我没必要先聚合每个类目的商品列表，这个操作本身就占用了不少资源。

reduceByKey的作用是把两个同类型的数据转化成单个同类型的数据，上两次迭代就是把商品列表以字符串的形式粘合在一起。

但就像maxBy函数的内部逻辑一样，每两个商品对比时我们只需要保留gmv较大的那个就行了。而reduceByKey可以先轻松的完成这个操作：

def getGoodsGmvTop(spark: SparkSession,
    order_goods_info: RDD[(String, String, String, String, String)], 
    brGoods2Catid: Broadcast[Map[String, String]]): RDD[(String, String)] ={
  order_goods_info.map(x => (x._2,1))
    .reduceByKey(_+_)
    .flatMap(x => (brGoods2Catid.value(x._1), (x._1,x._2)))
    .reduceByKey((x,y) => if (y._2 > x._2) y else x)
    .map(x => (x._1,x._2._1))
}

我是小结
最后一次迭代后的函数看上去就顺眼多了，三次迭代也让我对reduceByKey这个重要的函数有了更深的理解。这次改造后，顺势对原先的算法进行了大面积的优化，算法的运行时间缩短了不少。

另外，从这三次迭代，我总结出两点值得反复提醒自己的准则：

1. 写新算法或重构别人代码时，首先要清楚这一步我们的目的是什么。
2. 不要做不需要做的事