spark 例子count(distinct 字段)

---

例子描述：

有個網站訪問日志，有4個字段：（用戶id，用戶名，訪問次數，訪問網站）

需要統計：

1.用戶的訪問總次數去重

2.用戶一共訪問了多少種不同的網站

這里用sql很好寫

select id,name,count(distinct url) from table group by id,name

其實這個題目是繼官方和各種地方講解聚合函數（aggregate）的第二個例子，第一個例子是使用aggregate來求平均數。

我們先用簡易版來做一遍，后續我更新一份聚合函數版

原始數據：

id1,user1,2,http://www.baidu.com
id1,user1,2,http://www.baidu.com
id1,user1,3,http://www.baidu.com
id1,user1,100,http://www.baidu.com
id2,user2,2,http://www.baidu.com
id2,user2,1,http://www.baidu.com
id2,user2,50,http://www.baidu.com
id2,user2,2,http://www.sina.com

結果數據：

((id1,user1),4,1)
((id2,user2),4,2)

---

代碼片段：

val sparkConf = new SparkConf().setAppName("DisFie").setMaster("local")
val sc = new SparkContext(sparkConf)
 
val source = Source.fromFile("C:\\10.txt").getLines.toArray
val RDD0 = sc.parallelize(source)

RDD0
  .map {
    lines =>
      val line = lines.split(",")
      ((line(0), line(1)), (1, line(3)))
  }
  .groupByKey()
  .map {
    case (x, y) =>
      val(n,url) = y.unzip
    (x,n.size,url.toSet.size)
  }
  .foreach(println)