038 spark中使用sparksql对日志进行分析（属于小案例）

一：使用sparksql开发

1.sparksql开发的两种方式

　　HQL：SQL语句开发

　　　　eq : sqlContext.sql("xxxx")

　　DSL : sparkSql中DataFrame的API调用方式

　　　　eq：val df=sqlContext.xxx

　　　　　　 df.select("number")

二：HQL的开发案例

1.新建目录上传日志

　　

2.开启服务

　　

三：书写程序

1.描述

　　这个程序一共包括两个部分。

　　所以写的是两个程序。

2.程序一：对日志的描述--ApacheAccessLog

package com.ibeifeng.bigdata.spark.log

import scala.util.matching.Regex

/**
   * 64.242.88.10 - - [07/Mar/2004:16:05:49 -0800] "GET /twiki/bin/edit/Main/Double_bounce_sender?topicparent=Main.ConfigurationVariables HTTP/1.1" 401 12846
   * Created by ibf on 01/15.
   */
case class ApacheAccessLog(
                             ipAddress: String, // IP地址
                             clientId: String, // 客户端唯一标识符
                             userId: String, // 用户唯一标识符
                             serverTime: String, // 服务器时间
                             method: String, // 请求类型/方式
                             endpoint: String, // 请求的资源
                             protocol: String, // 请求的协议名称
                             responseCode: Int, // 请求返回值：比如：200、401
                             contentSize: Long // 返回的结果数据大小
                           )

/**
   * 64.242.88.10 - - [07/Mar/2004:16:05:49 -0800] "GET /twiki/bin/edit/Main/Double_bounce_sender?topicparent=Main.ConfigurationVariables HTTP/1.1" 401 12846
   * Created by ibf on 01/15.
   * 提供一些操作Apache Log的工具类供SparkCore使用
   */
object ApacheAccessLog {
   // Apache日志的正则
   val PARTTERN: Regex =
   """^(S+) (S+) (S+) [([w:/]+s[+-]d{4})] "(S+) (S+) (S+)" (d{3}) (d+)""".r

   /**
     * 验证一下输入的数据是否符合给定的日志正则，如果符合返回true；否则返回false
     *
     * @param line
     * @return
     */
   def isValidateLogLine(line: String): Boolean = {
     val options = PARTTERN.findFirstMatchIn(line)

     if (options.isEmpty) {
       false
     } else {
       true
     }
   }

   /**
     * 解析输入的日志数据
     *
     * @param line
     * @return
     */
   def parseLogLine(line: String): ApacheAccessLog = {
     if (!isValidateLogLine(line)) {
       throw new IllegalArgumentException("参数格式异常")
     }

     // 从line中获取匹配的数据
     val options = PARTTERN.findFirstMatchIn(line)

     // 获取matcher
     val matcher = options.get

     // 构建返回值
     ApacheAccessLog(
       matcher.group(1), // 获取匹配字符串中第一个小括号中的值
       matcher.group(2),
       matcher.group(3),
       matcher.group(4),
       matcher.group(5),
       matcher.group(6),
       matcher.group(7),
       matcher.group(8).toInt,
       matcher.group(9).toLong
     )
   }
 }

3.程序二：针对需求进行--LogAnalysis

package com.ibeifeng.bigdata.spark.log

import com.ibeifeng.bigdata.spark.core.ApacheAccessLog
import org.apache.spark.sql.{DataFrame, SQLContext}
import org.apache.spark.{SparkContext, SparkConf}
/**
 * Created by Administrator on 2017/4/25.
 */
object LogAnalysis {
  def main(args: Array[String]):Unit={
    //sqlContext
    val conf=new SparkConf()
      .setMaster("local[*]")
      .setAppName("log-analysis-sparksql")
    val sc=SparkContext.getOrCreate(conf)
    val sqlContext=new SQLContext(sc)
    import sqlContext.implicits._                //如果不写，下面的转换不成功

    //transform
    val path="/spark/logs/input"
    val rdd=sc.textFile(path)
    val apacheAccessDataFrame=rdd
      .filter(line=>ApacheAccessLog.isValidateLogLine(line))
      .map(line => {
        ApacheAccessLog.parseLogLine(line)
    }).toDF()                                    //rdd转换为DataFrame

    //register temptable
    apacheAccessDataFrame.registerTempTable("log_analysis_temp_table")
    sqlContext.sql("select * from log_analysis_temp_table limit 1").show()

    //需求一：求contentSize的平均值，最大值以及最小值
    val resultDataFrame1=sqlContext.sql(
      """
        |SELECT
        |AVG(contentSize) as avg_contentSize,
        |MAX(contentSize) as max_contentSize,
        |MIN(contentSize) as min_contentSize
        |FROM log_analysis_temp_table
      """.stripMargin)
    resultDataFrame1.show()

    //save                                         //save as HDFS
    val resultRdd=resultDataFrame1.map(row=>{
      val avgSize=row.getAs[Double]("avg_contentSize")
      val minSize=row.getAs[Long]("min_contentSize")
      val maxSize=row.getAs[Long]("max_contentSize")
      (avgSize,minSize,maxSize)
    })
    resultRdd.saveAsTextFile(s"/spark/logs/output/sql_${System.currentTimeMillis()}")

    //需求二：求各个返回值出现的数据个数
    val resultDataFrame2=sqlContext.sql(
    """
      |SELECT
      |responseCode AS code,
      |COUNT(1) AS count
      |FROM log_analysis_temp_table
      |GROUP BY responseCode
    """.stripMargin
    )
    resultDataFrame2.show()

    //需求三：求访问次数大于N的IP地址，并对黑名单进行限制
    val blackIP=Array("200-55-104-193.ds1.prima.net.ar","10.0.0.153","208-38-57-205.ip.cal.radiant.net")
    val N=10
    val resultDataFrame3=sqlContext.sql(
    s"""
      |SELECT
      |ipAddress AS ip,
      |COUNT(1) AS count
      |FROM log_analysis_temp_table
      |WHERE not(ipAddress in(${blackIP.map(ip=>s"'${ip}'").mkString(",")}))
      |GROUP BY ipAddress
      |HAVING count>${N}
    """.stripMargin)
    resultDataFrame3.show()

    //需求四：求访问次数最多的前k个endpoint的值
    val k=10
    val resultDataFrame4=sqlContext.sql(
    s"""
       |SELECT
       |  t.endpoint,
       |  t.count
       |FROM(
       |SELECT
       |  endpoint,
       |  COUNT(1) AS count
       |FROM log_analysis_temp_table
       |GROUP BY endpoint) t
       |ORDER BY t.count DESC
       |limit ${k}
     """.stripMargin)
    resultDataFrame4.show()
  }
}

4.运行结果