about云日志分析项目准备11-1：spark streaming+spark sql 实现业务

接上篇：about云日志分析项目准备11：spark streaming 接收 flume 监控目录的日志文件
这篇主要讲解如何使用spark streaming在本地实现业务逻辑

一、工程结构

整个工程分为以下几个模块
business：业务代码
conf：配置信息
preprocessing：日志预处理过程
util：工具类
Run：启动程序


二、Run

这是一个入口类，包含了spark streaming程序的启动和结果输出相关的程序

1. import java.text.SimpleDateFormat
   
2. import java.util.{Calendar, Date}
   
3. 
   
4. import business.StaSQL
   
5. import conf.Config
   
6. import org.apache.log4j.{Level, Logger}
   
7. import org.apache.spark.sql.hive.HiveContext
   
8. import org.apache.spark.sql.types.{StringType, StructField, StructType}
   
9. import org.apache.spark.streaming.dstream.DStream
   
10. import org.apache.spark.streaming.{Seconds, StreamingContext}
    
11. import org.apache.spark.{SparkConf, SparkContext}
    
12. import preprocessing.PreProcessing
    
13. import util.Tools
    
14. 
    
15. /**
    
16.   * 入口类
    
17.   */
    
18. object Run {
    
19.   def main(args: Array[String]): Unit = {
    
20. 
    
21.     Logger.getRootLogger.setLevel(Level.WARN)
    
22. 
    
23.     val date = new Date()
    
24.     val calendar = Calendar.getInstance()
    
25.     calendar.add(Calendar.DATE, -1)
    
26.     val total_time = calendar.getTime()
    
27.     val formatter = new SimpleDateFormat("yyyyMMdd")
    
28.     var dateString = formatter.format(total_time)
    
29. 
    
30.     // 创建SparkConf对象，并指定AppName和Master
    
31.     val conf = new SparkConf()
    
32.       .setAppName("StartProgram")
    
33.       .setMaster("local[2]")
    
34. 
    
35.     val sc = new SparkContext(conf)
    
36.     val sqlContext = new HiveContext(sc)
    
37.     // 创建StreamingContext对象
    
38.     val ssc = new StreamingContext(sc, Seconds(10))
    
39. 
    
40.     // 本地读取日志文件
    
41.     val dir = "D:\\work\\aboutyun_log_analysis\\log\"
    
42.     val raw_log: DStream[String] = ssc.textFileStream(dir)
    
43. 
    
44.     val pp = PreProcessing(ssc, sqlContext, raw_log)
    
45.     val log_dstream = pp.transformLogFormat()
    
46. 
    
47.     val schemaLogString = "date c_ip user_agent url url_part tid fid navigation"
    
48.     val schemaLog =
    
49.       StructType(
    
50.         schemaLogString.split(" ")
    
51.           .map(fieldName => StructField(fieldName, StringType)))
    
52. 
    
53.     // 用于统计业务的spark sql封装对象
    
54.     val staSql = StaSQL(dateString)
    
55. 
    
56.     // 针对每一个RDD，转为DataFrame，并使用spark sql来计算结果，并将结果输出到mysql中
    
57.     log_dstream.foreachRDD{rdd =>
    
58.       println("-------------------------------------")
    
59.       sqlContext.createDataFrame(rdd, schemaLog).registerTempTable("user_behavior")
    
60. //      sqlContext.sql("select * from user_behavior").show(20, false)
    
61.       sqlContext.sql(staSql.sql_count_per_ip_pvs).rdd.foreachPartition(iter =>
    
62.         Tools.rdd2mysql(Config.aboutyun_ip_sta_table, Config.aboutyun_ip_sta_table_schema, iter))
    
63.       sqlContext.sql(staSql.sql_count_top_50_tid).rdd.foreachPartition(iter =>
    
64.         Tools.rdd2mysql(Config.aboutyun_hot_post_sta_table, Config.aboutyun_hot_post_sta_table_schema, iter))
    
65.       sqlContext.sql(staSql.sql_count_module_pvs).rdd.foreachPartition(iter =>
    
66.         Tools.rdd2mysql(Config.aboutyun_module_sta_table, Config.aboutyun_module_sta_table_schema, iter))
    
67.       sqlContext.sql(staSql.sql_count_navigation).rdd.foreachPartition(iter =>
    
68.         Tools.rdd2mysql(Config.aboutyun_navigation_sta_table, Config.aboutyun_navigation_sta_table_schema, iter))
    
69.       sqlContext.sql(staSql.sql_count_in_and_out_station_request).rdd.foreachPartition(iter =>
    
70.         Tools.rdd2mysql(Config.aboutyun_station_sta_table, Config.aboutyun_station_sta_table_schema, iter))
    
71.       sqlContext.sql(staSql.sql_count_per_ip_per_url_pvs).rdd.foreachPartition(iter =>
    
72.         Tools.rdd2mysql(Config.aboutyun_suspicious_ip_sta_table, Config.aboutyun_suspicious_ip_sta_table_schema, iter))
    
73.     }
    
74. 
    
75.     ssc.start()
    
76.     ssc.awaitTermination()
    
77. 
    
78.   }
    
79. 
    
80. }
    

复制代码

三、conf包

conf包中包括了Config伴生类，其中包含了论坛版块名称和id的映射关系、mysql的配置信息、以及输出结果的表名和字段名。


四、preprocessing包

preprocessing包中包括了spark streaming程序的日志格式转化、以及一些辅助临时表的注册。

1. class PreProcessing(ssc: StreamingContext, sqlContext: HiveContext, raw_log: DStream[String]) {
   
2. 
   
3.   // 注册一些后面需要用到的关联表
   
4.   def registerHelperTable(): Unit ={
   
5.     val sc = ssc.sparkContext
   
6.     val rddModuleMap = sc.parallelize(Config.moduleMap)
   
7.       .map(s => Row.fromSeq(s.split(","))) // 构造板块名称和id映射关系
   
8. 
   
9.     val schemaModuleMapString = "module id"
   
10.     val schemaModuleMap =
    
11.       StructType(
    
12.         schemaModuleMapString.split(" ")
    
13.           .map(fieldName => StructField(fieldName, StringType)))
    
14. 
    
15.     sqlContext.createDataFrame(rddModuleMap, schemaModuleMap).registerTempTable("module_map")
    
16.   }
    
17. 
    
18.   // 将原始的日志格式转为需要用到的格式
    
19.   def transformLogFormat(): DStream[Row] ={
    
20. 
    
21.     registerHelperTable()
    
22.     val url_regex = "https|http://.*(?=\\))".r  // 匹配出后面是")"的一个url即为包含在user-agent中的url
    
23.     val tid_regex = "((?<=tid=)|(?<=thread-))\\d+".r  // 匹配出前面是"tid="或是"thread-"的数字记为帖子id
    
24.     val fid_regex = "((?<=fid=)|(?<=forum-))\\d+".r  // 匹配出前面是"fid="或是"formu-"的数字记为板块id
    
25. 
    
26.     val log_dstream = raw_log
    
27.       .filter(text => !text.contains("#")) // 去除掉带有注释的行
    
28.       .map(text => text.split(" ")) // 对每一行使用空格来切分
    
29.       .filter(array => array(3).contains(".php")) // 保留带有".php"的内容
    
30.       .map { array =>
    
31.       val url = if (array(6) != "-") array(6) else url_regex.findFirstIn(array(5)).getOrElse("")
    
32.       val tid = tid_regex.findFirstIn(url).getOrElse("")
    
33.       val fid = fid_regex.findFirstIn(url).getOrElse("")
    
34.       // println(url)
    
35.       val requestParameter =  url match {
    
36.         case "http://www.aboutyun.com/home.php" => ""
    
37.         case _ =>
    
38.           if (url.split("\\?").length >1) url.split("\\?")(1) else "其他"
    
39.       }
    
40.       // if (url.split("\\?").length == 1 && url.eq("http://www.aboutyun.com/home.php")) "" else url.split("\\?")(1)
    
41.       val navigation = requestParameter match {
    
42.         case "mod=guide" => "导读"
    
43.         case "mod=space&do=blog" => "博客"
    
44.         case "mod=collection" => "专题"
    
45.         case "mod=space&do=share" => "阅读分享"
    
46.         case "" => "浏览"
    
47.         case _ => "其他"
    
48.       }
    
49.       Row(array(0) + " " + array(1), array(4), array(5), url, array(3), tid, fid, navigation)
    
50.     }
    
51.     log_dstream
    
52.   }
    
53.   
    
54. }

复制代码

五、 business包

business包中的StaSQL包含了真正计算业务用到的Spark SQL语句。



六、util包

util包中包括了我们用到的输出结果表的建表语句和将rdd写入到mysql。

1. package util
   
2. 
   
3. import java.sql.{Connection, DriverManager, PreparedStatement}
   
4. 
   
5. import conf.Config
   
6. 
   
7. /**
   
8.   * 封装的工具类
   
9.   */
   
10. object Tools {
    
11. 
    
12.   // 将rdd的内容输出到mysql中
    
13.   def rdd2mysql(tableName: String, schema: String, iterator: Iterator[org.apache.spark.sql.Row]): Unit ={
    
14. 
    
15.     def getFieldAndTypeTuple(schema: String): Array[(String, String)] = {
    
16.       schema.split(",").map{s =>
    
17.         val Array(field, fieldType) = s.trim.split(":")
    
18.         (field, fieldType)
    
19.       }
    
20.     }
    
21. 
    
22.     val iterArray = iterator.toArray
    
23.     val fieldAndTypeTuple: Array[(String, String)] = getFieldAndTypeTuple(schema)
    
24.     val fields = fieldAndTypeTuple.map(_._1).mkString(",")
    
25.     val values = (for (i <- 0 until fieldAndTypeTuple.length) yield "?").mkString(",")
    
26. 
    
27.     var conn: Connection = null
    
28.     var ps: PreparedStatement = null
    
29.     val sql = "insert into %s (%s) values (%s)".format(tableName, fields, values)
    
30.     try {
    
31. //      conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/recommenddb", "root", "nopassword")
    
32.       conn = DriverManager.getConnection(Config.mysqlUrl, Config.mysqlUser, Config.mysqlPassword)
    
33.       conn.setAutoCommit(false)
    
34.       ps = conn.prepareStatement(sql)
    
35.       // row即为每个rdd中的row
    
36.       //      println(sql)
    
37.       //      println(iterArray.mkString("@"))
    
38.       for (row <- iterArray){
    
39.         ps.clearBatch()
    
40.         for (i <- 0 until row.length){
    
41.           val fieldType = fieldAndTypeTuple(i)._2
    
42.           if ( fieldType == "string"){
    
43.             ps.setString(i + 1, row(i).toString)
    
44.           }else if (fieldType == "int"){
    
45.             ps.setInt(i + 1, row(i).toString.toInt)
    
46.           }else if (fieldType == "double"){
    
47.             ps.setDouble(i + 1, row(i).toString.toDouble)
    
48.           }
    
49. 
    
50.         }
    
51.         ps.addBatch()
    
52.         ps.executeBatch()
    
53.       }
    
54. 
    
55.       conn.commit()
    
56. 
    
57.     } catch {
    
58.       case e: Exception => println(e)
    
59.     } finally {
    
60.       if (ps != null) {
    
61.         ps.close()
    
62.       }
    
63.       if (conn != null) {
    
64.         conn.close()
    
65.       }
    
66.     }
    
67. 
    
68.   }
    
69. }

复制代码

注：完整代码见：链接：http://pan.baidu.com/s/1bp2AuuF 密码：cn7a