网站用户行为分析项目之会话切割（四）=> 代码重构

墨蓝 2021-09-23 13:48 328阅读 0赞

文章目录

  • 0x00 文章内容
  • 0x01 实现输出代码的重构
            1. 抽离输出代码
            1. 重构输出路径
            1. 重构输出文件类型
  • 0x02 校验结果
            1. 生成不同的输出文件类型
  • 0xFF 总结

0x00 文章内容

  1. 实现输出代码的重构
  2. 校验结果

0x01 实现输出代码的重构

1. 抽离输出代码

a. 因为SessionCutETL里的main方法写了比较多的代码，此时我们可以将第6步骤的输出代码进行抽离，全选，选中Refactor=>Extract=>Method
在这里插入图片描述
b. 我们这里选择第一个在这里插入图片描述
c. 填写方法名，点击OK
在这里插入图片描述
即可发现已经抽离出了方法。

2. 重构输出路径

a. 下面的代码中路径都是写死的，而且出现了共同的路径，我们可以进行统一

  1. val trackerLogOutputPath = "data/output/trackerLog"
  2.     val trackerSessionOutputPath = "data/output/trackerSession"

修改如下：

  1. writeOutputData(sc,"data/output", parsedLogRDD, cookieLabeledSessionRDD)
  2.     private def writeOutputData(sc: SparkContext, baseOutputPath: String, parsedLogRDD: RDD[TrackerLog], cookieLabeledSessionRDD: RDD[TrackerSession]) = { 
  3.     val trackerLogOutputPath = s"${baseOutputPath}/trackerLog"
  4.     val trackerSessionOutputPath = s"${baseOutputPath}/trackerSession"

我们暂且这样重构先，此时需要重新执行一下代码，看一下改后是否能执行。

d. 执行验证，执行会报错，表示路径的文件已经存在，此时可以手动删除再执行，这里下一步是在代码中实现删除。

  1. Exception in thread "main" org.apache.hadoop.mapred.FileAlreadyExistsException: Output directory data/output/trackerLog already exists

e. 添加判断路径代码

  1. val fileSystem = FileSystem.get(sc.hadoopConfiguration)
  2.     val path = new Path(trackerLogOutputPath)
  3.     if (fileSystem.exists(path)) { 
  4.       fileSystem.delete(path, true)
  5.     }

写完之后再抽离出去，取名deletePathIfExists方法。

f. 此时再重新执行代码，发现代码没有报错，也能得到想要的结果。

3. 重构输出文件类型

目前的情况是以Parquet的形式保存着，如果此时如果需求发生变化了，或者有其他格式的需求，如保存成TextFile格式。就要重新改代码了，如果需求又变了，或者写成其他组件，又要重新改，重新然后打包，重新然后上传，这样非常麻烦。试想，如果此时将需要修改的代码抽象成一个接口，就会大大方便了。

a. 本优化内容比较多，涉及两个类，代码量及过程比较多：
在这里插入图片描述
b. OutputComponent完整代码如下：

  1. package com.shaonaiyi.session
  3. import com.shaonaiyi.spark.session.{ TrackerLog, TrackerSession}
  4. import org.apache.hadoop.fs.{ FileSystem, Path}
  5. import org.apache.parquet.avro.{ AvroParquetOutputFormat, AvroWriteSupport}
  6. import org.apache.spark.SparkContext
  7. import org.apache.spark.rdd.RDD
  9. /** * @Auther: shaonaiyi@163.com * @Date: 2019/12/30 21:09 * @Description: 抽象输出文件组件 */
  10. trait OutputComponent { 
  12.   def writeOutputData(sc: SparkContext, baseOutputPath: String,
  13.                       parsedLogRDD: RDD[TrackerLog],
  14.                       cookieLabeledSessionRDD: RDD[TrackerSession]) = { 
  15.     deletePathIfExists(sc, baseOutputPath)
  16.   }
  18.   private def deletePathIfExists(sc: SparkContext, trackerLogOutputPath: String) = { 
  19.     val fileSystem = FileSystem.get(sc.hadoopConfiguration)
  20.     val path = new Path(trackerLogOutputPath)
  21.     if (fileSystem.exists(path)) { 
  22.       fileSystem.delete(path, true)
  23.     }
  24.   }
  26. }
  28. object OutputComponent { 
  29.   def fromOutputFileType(fileType: String) = { 
  30.     if (fileType.equals("parquet")) { 
  31.       new ParquetFileOutput
  32.     } else { 
  33.       new TextFileOutput
  34.     }
  35.   }
  36. }
  38. /** * 写Parquet格式文件 */
  39. class ParquetFileOutput extends OutputComponent { 
  40.   override def writeOutputData(sc: SparkContext, baseOutputPath: String,
  41.                                parsedLogRDD: RDD[TrackerLog],
  42.                                cookieLabeledSessionRDD: RDD[TrackerSession]): Unit = { 
  44.     super.writeOutputData(sc, baseOutputPath, parsedLogRDD, cookieLabeledSessionRDD)
  46.     //6、保存数据
  47.     //6.1、保存TrackerLog，对应的是parsedLogRDD
  48.     val trackerLogOutputPath = s"${baseOutputPath}/trackerLog"
  50.     AvroWriteSupport.setSchema(sc.hadoopConfiguration, TrackerLog.SCHEMA$)
  51.     parsedLogRDD.map((null, _)).saveAsNewAPIHadoopFile(trackerLogOutputPath,
  52.       classOf[Void], classOf[TrackerLog], classOf[AvroParquetOutputFormat[TrackerLog]]
  53.     )
  54.     //6.2、保存TrackerSession，对应的是cookieLabeledSessionRDD
  55.     val trackerSessionOutputPath = s"${baseOutputPath}/trackerSession"
  57.     AvroWriteSupport.setSchema(sc.hadoopConfiguration, TrackerSession.SCHEMA$)
  58.     cookieLabeledSessionRDD.map((null, _)).saveAsNewAPIHadoopFile(trackerSessionOutputPath,
  59.       classOf[Void], classOf[TrackerSession], classOf[AvroParquetOutputFormat[TrackerSession]]
  60.     )
  61.   }
  63. }
  65. /** * 写TextFile格式文件 */
  66. class TextFileOutput extends OutputComponent { 
  67.   override def writeOutputData(sc: SparkContext, baseOutputPath: String,
  68.                                parsedLogRDD: RDD[TrackerLog],
  69.                                cookieLabeledSessionRDD: RDD[TrackerSession]): Unit = { 
  71.     super.writeOutputData(sc, baseOutputPath, parsedLogRDD, cookieLabeledSessionRDD)
  73.     //6、保存数据
  74.     //6.1、保存TrackerLog，对应的是parsedLogRDD
  75.     val trackerLogOutputPath = s"${baseOutputPath}/trackerLog"
  77.     parsedLogRDD.saveAsTextFile(trackerLogOutputPath)
  79.     //6.2、保存TrackerSession，对应的是cookieLabeledSessionRDD
  80.     val trackerSessionOutputPath = s"${baseOutputPath}/trackerSession"
  82.     cookieLabeledSessionRDD.saveAsTextFile(trackerSessionOutputPath)
  84.   }
  86. }

编写思路：先写OutputComponent接口，然后写ParquetFileOutput类继承OutputComponent，最后写一个伴生类OutputComponent以方便调用，最后进行代码优化，将deletePathIfExists抽离出来。

代码讲解：

  1. 此OutputComponent是一个Trait类型，属于接口抽象类，目的将输出格式接口化，原本是只有一个Parquet格式的，现在再添加了一个TextFile格式，实现的功能其实就是与输出Parquet格式的代码相类似。ParquetFileOutput与TextFileOutput均继承此类，所以需要实现writeOutputData方法，写好之后，需要再写一个伴生类来调用，并且判断输入的是哪种类型；除此之外，还简化了deletePathIfExists方法，统一用super进行了了调用。
  2. saveAsNewAPIHadoopFile需要的参数是key-value类型，所以需要转成key-value先。

c. SessionCutETL完整代码如下：

  1. package com.shaonaiyi.session
  3. import com.shaonaiyi.spark.session.{ TrackerLog, TrackerSession}
  4. import org.apache.spark.rdd.RDD
  5. import org.apache.spark.{ SparkConf, SparkContext}
  7. /** * @Auther: shaonaiyi@163.com * @Date: 2019/9/12 10:09 * @Description: 会话切割的程序主入口 */
  8. object SessionCutETL { 
  10.   private val logTypeSet = Set("pageview", "click")
  12.   def main(args: Array[String]): Unit = { 
  14.     var conf = new SparkConf()
  15.     conf.setAppName("SessionCutETL")
  16.     conf.setMaster("local")
  17.     conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
  18.     var sc = new SparkContext(conf)
  20.     //网站域名标签数据，此处只是演示，其实可以存放在数据库里
  21.     val domainLabelMap = Map(
  22.       "www.baidu.com" -> "level1",
  23.       "www.taobao.com" -> "level2",
  24.       "jd.com" -> "level3",
  25.       "youku.com" -> "level4"
  26.     )
  28.     //广播
  29.     val domainLabelMapB = sc.broadcast(domainLabelMap)
  31. // sc.setLogLevel("ERROR")
  33.     // 1、加载日志源数据
  34.     val rawRDD: RDD[String] = sc.textFile("data/rawdata/visit_log.txt")
  35. // rawRDD.collect().foreach(println)
  37.     //2、解析rawRDD中每一行日志源数据
  38. // val parsedLogRDD: RDD[Option[TrackerLog]] = rawRDD.map( line => RawLogParserUtil.parse(line))
  39. // val parsedLogRDD1: RDD[TrackerLog] = rawRDD.flatMap( line => RawLogParserUtil.parse(line))
  40.     val parsedLogRDD: RDD[TrackerLog] = rawRDD.flatMap( line => RawLogParserUtil.parse(line))
  41.     .filter(trackerLog => logTypeSet.contains(trackerLog.getLogType.toString))
  42. // parsedLogRDD.collect().foreach(println)
  43. // parsedLogRDD1.collect().foreach(println)
  45.     //3、按照cookie进行分组
  46.    val cookieGroupRDD: RDD[(String, Iterable[TrackerLog])] = parsedLogRDD.groupBy(trackerLog => trackerLog.getCookie.toString)
  47. // cookieGroupRDD.collect().foreach(println)
  49.     //4、按user进行分组
  50.     val sessionRDD: RDD[(String, TrackerSession)] = cookieGroupRDD.flatMapValues {  case iter =>
  52.       //处理每个user的日志
  53.       val processor = new OneUserTrackerLogsProcessor(iter.toArray)
  54.         processor.buildSessions(domainLabelMapB.value)
  55.     }
  57.     //5、给会话的cookie打标签
  58.     val cookieLabelRDD: RDD[(String, String)] = sc.textFile("data/cookie_label.txt").map {  case line =>
  59.       val temp = line.split("\\|")
  60.       (temp(0), temp(1)) // (cookie, cookie_label)
  61.     }
  63.     val joinRDD: RDD[(String,(TrackerSession, Option[String]))] = sessionRDD.leftOuterJoin(cookieLabelRDD)
  65.     val cookieLabeledSessionRDD: RDD[TrackerSession] = joinRDD.map { 
  66.       case (cookie, (session, cookieLabelOpt)) =>
  67.         if (cookieLabelOpt.nonEmpty) { 
  68.           session.setCookieLabel(cookieLabelOpt.get)
  69.         } else { 
  70.           session.setCookieLabel("-")
  71.         }
  72.         session
  73.     }
  75.     //text & parquet
  76.     OutputComponent.fromOutputFileType("text").writeOutputData(sc,"data/output", parsedLogRDD, cookieLabeledSessionRDD)
  78.     sc.stop()
  79.   }
  81. }

代码讲解：删除原先的输出格式的代码，用OutputComponent调用来实现。

0x02 校验结果

1. 生成不同的输出文件类型

a. 执行SessionCutETL类，查看文件的输出类型，此时生成了TextFile格式文件。
在这里插入图片描述
b. 修改此行代码的”text”为parquet，重新执行，查看结果，此时生成了Parquet格式文件。

  1. //text & parquet
  2.     OutputComponent.fromOutputFileType("parquet").writeOutputData(sc,"data/output", parsedLogRDD, cookieLabeledSessionRDD)

在这里插入图片描述

0xFF 总结

  1. 代码重构是一项非常重要的技能，增强代码的可读性。
  2. 下一篇文章还会继续优化，请留意本博客，关注、评论、加点赞。
  3. 网站用户行为分析项目系列：
    网站用户行为分析项目之会话切割（一）
    网站用户行为分析项目之会话切割（二）
    网站用户行为分析项目之会话切割（三）
    网站用户行为分析项目之会话切割（四）待补充
    网站用户行为分析项目之会话切割（五）待补充

作者简介：邵奈一
全栈工程师、市场洞察者、专栏编辑
| 公众号 | 微信 | 微博 | CSDN | 简书 |

福利：
邵奈一的技术博客导航
邵奈一 原创不易，如转载请标明出处。

发表评论

表情：
评论列表 （有 0 条评论，328人围观）

还没有评论，来说两句吧...

相关阅读

    相关 用户行为分析系统

    刚刚做完提交代码，等待测试的反馈，怀着激动的心情，先看看效果，代码过段时间会放到github上。已经在这儿：[https://github.com/awhen/Big-data

    电玩女神电玩女神/ 0 赞/ 337 阅读