Spark的reduceByKey方法使用

一、需求

在ODPS上���们有如下数据：

id	category_id	attr_id	attr_name	attr_value
205348	10000046	2	最优粘度	["0W-40"]
205348	10000046	1	基础油类型	["全合成"]
205348	10000046	3	级别	["BMW Longlife 01"]

我们希望得到的结果如下：

(205348, 10000046, "基础油类型：全合成\n最优粘度：0W-40\n级别：BMW Longlife 01\n")

需求解读：

需要将(id, category_id)作为key，然后将(attr_id, attr_name, attr_value)进行reduce操作，在reduce之后的数据中对attr_id进行排序，再将attr_name和attr_value合并在一起。

二、reduce操作之字符串方式

这个是最简单的方式，大致思路如下：

首先，将(id, category_id)作为key。

然后，将attr_id、attr_name、attr_value合并成一个字符串attr_info：attr_id + "#" + attr_name + "#" + attr_value，然后attr_info再通过"&"进行合并。

示例代码如下：

xx.map{case(id, category_id, attr_id, attr_name, attr_value) => ((id, category_id), attr_id + "#" + attr_name + "#" + attr_value)}.reduceByKey(_ + "&" + _, 100)

然后在接下来的流程中首先split("#")得到不同的attr信息，再通过split("&")得到不同的attr的列信息。这就要求attr_id，attr_name，attr_value中不能包含"#"和"&"字符串。

所以这种方式有缺陷，就是当attr_id，attr_name，attr_value包含了"#"和"&"字符串时需要先replace一下，这样就改变了原数据的值。

三、reduce操作之列表方式

这种方式相对复杂一点，需要对输入数据进行预处理，但是逻辑清晰。

输入数据中(id, category_id)是key保持不变，(item_id, item_name, item_value)是一组tuple。

reduce操作会在同一个partition中，不同的partition之间进行数据合并，这要求数据的输入、输出类型保持不变。

我们的初步想法：将item_id, item_name, item_value分别放到3个列表中，合并时就是列表之间的合并，合并完毕后使用时只需要遍历列表即可。

因为reduce操作的输入、输出类型不能变化，所以先放item_id, item_name, item_value初始化为一个列表，然后再进行列表之间的合并。

示例代码如下：

xx.map{case(id, category_id, attr_id, attr_name, attr_value) =>   val itemIdList = new ArrayList[Long]()  itemIdList.add(attr_id)  val itemNameList = new ArrayList[String]()  itemNameList.add(attr_name)  val itemValueList = new ArrayList[String]()  itemValueList.add(attr_value)  ((id, category_id), (itemIdList, itemNameList, itemValueList))}.reduceByKey((x, y) => {  val itemIdList = new ArrayList[Long]()  for(i