spark-distinct算子

上一篇已经简单说过了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，谢谢！