上一篇已经简单说过了sortBy算子传送门。今天，我们来了解一下distinct算子，他们也是Transformation算子
模拟一个场景，来讲各个Transformation的API
在这里我们模拟一个统计网站页面浏览情况的需求。
业务场景
我们这里使用网站浏览日志作为业务场景，就用简单的三个字段页面埋点id、用户id、停留时长
业务场景
日志格式
假定网站浏览日志以HDFS形式通过我们的系统埋点到我们数据仓库中了，埋点日志的数据格式如下：
日志格式
假定我们代码中已经使用这样的格式，把日志记录到服务器上，并通过flume上传到了HDFS服务器上
页面id|用户id|停留时长
index|2|6
表示：用户id为2的用户访问了index页面，并停留了6秒钟
这次，我们包装一个实体类，BrowserLogInfo，字段分别为url（用户访问页面），userId(用户id)，time(停留时长)，来映射我们日志中的三个字段
下面接入正题，来看distinct算子
- distinct就是用来对数据进行去重的算子
去除数组中的重复数据
- 先直接上代码，来看使用方式吧
- 代码逻辑为，从日志中读取数据，获取到用户访问过的所有地址，然后去重，来看用户访问过的地址有哪些
直接上代码举栗子
package com.edu.spark.rdd.transformation
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
/**
* @description: .
* @author: XiaoCaoMei .
* @createdTime: 2019/12/21 16:41.
* @version: 1.0 .
*/
object LogProcess {
def main(args: Array[String]): Unit = {
val conf = new SparkConf()
/**
* 如果这个参数不设置，默认认为你运行的是集群模式
* 如果设置成local代表运行的是local模式
*/
conf.setMaster("local")
//设置任务名
conf.setAppName("EduSpark")
//创建SparkCore的程序入口
val sc = new SparkContext(conf)
//读取文件 生成RDD
val file : RDD[String] = sc.textFile("F:\\hdfs\\hello.log")
println(file.collect().toBuffer)
//把每一行数据按照 | 分割
val traceRdd : RDD[String] = file.map(s => {
//把每一行数据按照 | 分割
//注意这里 | 需要使用转义字符
var sArray: Array[String] = s.split("\\|")
//split函数的作用是 通过|分隔字符串返回数组
// sArray 数组为 url | userId | time
// 返回 url
sArray(0)
})
//去重之前打印
println(traceRdd.collect().toBuffer)
//使用distinct算子去重，最后得到所有访问过的url
val distinctRdd : RDD[String] = traceRdd.distinct()
println(distinctRdd.collect().toBuffer)
sc.stop()
}
}
- 结果如下：
-
去重前打印
去重前打印 -
去重后打印
去重后打印
spark中的distinct算子就简单给大家说到这里，欢迎大家来交流，指出文中一些说错的地方，让我加深认识，愿大家没有bug，谢谢！
