Spark SQL 筆記(10)——實戰網站日誌分析（1）
阿新 • • 發佈：2018-12-21
1 使用者行為日誌介紹
1.1 行為日誌生成方法
- Nginx
- Ajax
1.2 日誌內容
- 訪問的系統屬性：作業系統、瀏覽器
- 訪問特徵：點選的 url、從哪個url 跳轉過來的（referer）、頁面停留時間
- 訪問資訊： session_id, 訪問ip,
2 離線資料處理架構
- 資料採集： Flume: web日誌寫入到 HDFS
- 資料清洗：Spark,hive,mapreduce,清洗後可以存放到HDFS
- 資料處理：按照需求進行相應的業務統計分析
- 處理結果入庫：存放到 RDBMS,NoSQL
- 資料視覺化：Echarts,HUE, Zeppelin
3 需求
- 主站最受歡迎的課程/手記 Top N訪問次數；
- 按照地市統計最受歡迎的 Top N 課程；
- 根據IP地址提取出城市資訊；
- 視窗函式在 Spark SQL 中的使用；
- 按照流量統計最受歡迎的 Top N;
4 原始日誌清洗
原始記錄案例
183.162.52.7 - - [10/Nov/2016:00:01:02 +0800] "POST /api3/getadv HTTP/1.1" 200 813 "www.imooc.com" "-" cid=0×tamp=1478707261865&uid=2871142&marking=androidbanner&secrect=a6e8e14701ffe9f6063934780d9e2e6d&token=f51e97d1cb1a9caac669ea8acc162b96 "mukewang/5.0.0 (Android 5.1.1; Xiaomi Redmi 3 Build/LMY47V),Network 2G/3G" "-" 10.100.134.244:80 200 0.027 0.027
擷取前 20000 條資料
$ head -20000 access.20161111.log >> access_20000.log
4.1 第一次清洗
打斷點
4.2 原始碼
DateUtils.scala
package com.weblog.cn
import java.util.{Date, Locale}
import org.apache.commons.lang3.time.FastDateFormat
/*
* 日期時間解析類
* */
object DateUtils {
/*
* SimpleDateFormat 是執行緒不安全的
* */
//輸入檔案時間格式 [ SparkStatFormatJob.scala
package com.weblog.cn
import org.apache.spark.sql.SparkSession
/*
*
* */
object SparkStatFormatJob {
def main(args: Array[String]): Unit = {
val spark = SparkSession.builder().appName(" SparkStatFormatJobApp")
.master("local[2]").getOrCreate()
val access = spark.sparkContext.textFile("file:///d://access_20000.log")
//access.take(10).foreach(println)
/*access.map(line =>{
val splits = line.split(" ")
val ip = splits(0)
ip
}).take(10).foreach(println)*/
/*access.map(line =>{
val splits = line.split(" ")
val ip = splits(0)
//[10/Nov/2016:00:01:02 +0800]
val time = splits(3)+" "+splits(4)
val url = splits(11).replaceAll("\"","")
//流量
val traffic = splits(9)
(ip,DateUtils.parse(time),url,traffic)
}).take(10).foreach(println)*/
access.map(line => {
val splits = line.split(" ")
val ip = splits(0)
//[10/Nov/2016:00:01:02 +0800]
val time = splits(3) + " " + splits(4)
val url = splits(11).replaceAll("\"", "")
//流量
val traffic = splits(9)
DateUtils.parse(time) + "\t" + url + "\t" + traffic + "\t" + ip
}).saveAsTextFile("file:///d://weblog")
}
}
結果
2016-11-10 00:01:02 - 813 183.162.52.7
2016-11-10 00:01:02 - 0 10.100.0.1
2016-11-10 00:01:02 http://www.imooc.com/code/1852 2345 117.35.88.11
2016-11-10 00:01:02 - 94 182.106.215.93
2016-11-10 00:01:02 - 0 10.100.0.1
2016-11-10 00:01:02 - 19501 183.162.52.7
2016-11-10 00:01:02 - 0 10.100.0.1
2016-11-10 00:01:02 - 2510 114.248.161.26
2016-11-10 00:01:02 - 633 120.52.94.105
2016-11-10 00:01:02 - 0 10.100.0.1
2016-11-10 00:01:02 - 94 112.10.136.45
2016-11-10 00:01:02 http://www.imooc.com/code/2053 331 211.162.33.31
2016-11-10 00:01:02 http://www.imooc.com/code/3500 54 116.22.196.70
2016-11-10 00:01:02 - 0 10.100.0.1
2016-11-10 00:01:02 - 125 113.47.86.12
2016-11-10 00:01:02 http://www.imooc.com/code/547 54 119.130.229.90
4.2 第二次清洗
- 使用 Spark SQL 解析訪問日誌
- 解析出課程編號、型別
- 根據IP解析出城市資訊
https://github.com/wzhe06/ipdatabase - 使用 Spark SQL 將訪問時間按天進行分割槽輸出
第一次清洗的結果
2017-05-11 14:09:14 http://www.imooc.com/video/4500 304 218.75.35.226
2017-05-11 15:25:05 http://www.imooc.com/video/14623 69 202.96.134.133
2017-05-11 07:50:01 http://www.imooc.com/article/17894 115 202.96.134.133
2017-05-11 02:46:43 http://www.imooc.com/article/17896 804 218.75.35.226
2017-05-11 09:30:25 http://www.imooc.com/article/17893 893 222.129.235.182
2017-05-11 08:07:35 http://www.imooc.com/article/17891 407 218.75.35.226
2017-05-11 19:08:13 http://www.imooc.com/article/17897 78 202.96.134.133
4.3 使用 github 開源專案獲取城市
https://github.com/wzhe06/ipdatabase
4.3.1 下載後解壓，然後編譯
4.3.2 安裝 jar 包到自己的Maven倉庫
mvn install:install-file -Dfile=d://ipdatabase-1.0-SNAPSHOT.jar -DgroupId=com.ggstar -DartifactId=ipdatabase -Dversion=1.0 -Dpackaging=jar
4.3.3 修改 pom 檔案
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.tzb.demo2</groupId>
<artifactId>SQLContext</artifactId>
<version>1.0</version>
<inceptionYear>2008</inceptionYear>
<properties>
<scala.version>2.11.8</scala.version>
<spark.version>2.1.0</spark.version>
</properties>
<repositories>
<repository>
<id>scala-tools.org</id>
<name>Scala-Tools Maven2 Repository</name>
<url>http://scala-tools.org/repo-releases</url>
</repository>
</repositories>
<pluginRepositories>
<pluginRepository>
<id>scala-tools.org</id>
<name>Scala-Tools Maven2 Repository</name>
<url>http://scala-tools.org/repo-releases</url>
</pluginRepository>
</pluginRepositories>
<dependencies>
<dependency>
<groupId>org.scala-lang</groupId>
<artifactId>scala-library</artifactId>
<version>${scala.version}</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.11</artifactId>
<version>${spark.version}</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-hive_2.11</artifactId>
<version>${spark.version}</version>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>5.1.38</version>
</dependency>
<dependency>
<groupId>com.ggstar</groupId>
<artifactId>ipdatabase</artifactId>
<version>1.0</version>
</dependency>
<dependency>
<groupId>org.apache.poi</groupId>
<artifactId>poi-ooxml</artifactId>
<version>3.14</version>
</dependency>
<dependency>
<groupId>org.apache.poi</groupId>
<artifactId>poi</artifactId>
<version>3.14</version>
</dependency>
</dependencies>
<build>
<sourceDirectory>src/main/scala</sourceDirectory>
<testSourceDirectory>src/test/scala</testSourceDirectory>
<plugins>
<plugin>
<groupId>org.scala-tools</groupId>
<artifactId>maven-scala-plugin</artifactId>
<executions>
<execution>
<goals>
<goal>compile</goal>
<goal>testCompile</goal>
</goals>
</execution>
</executions>
<configuration>
<scalaVersion>${scala.version}</scalaVersion>
<args>
<arg>-target:jvm-1.5</arg>
</args>
</configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-eclipse-plugin</artifactId>
<configuration>
<downloadSources>true</downloadSources>
<buildcommands>
<buildcommand>ch.epfl.lamp.sdt.core.scalabuilder</buildcommand>
</buildcommands>
<additionalProjectnatures>
<projectnature>ch.epfl.lamp.sdt.core.scalanature</projectnature>
</additionalProjectnatures>
<classpathContainers>
<classpathContainer>org.eclipse.jdt.launching.JRE_CONTAINER</classpathContainer>
<classpathContainer>ch.epfl.lamp.sdt.launching.SCALA_CONTAINER</classpathContainer>
</classpathContainers>
</configuration>
</plugin>
</plugins>
</build>
<reporting>
<plugins>
<plugin>
<groupId>org.scala-tools</groupId>
<artifactId>maven-scala-plugin</artifactId>
<configuration>
<scalaVersion>${scala.version}</scalaVersion>
</configuration>
</plugin>
</plugins>
</reporting>
</project>
4.3.4 測試ip地址解析
package com.weblog.cn
import com.ggstar.util.ip.IpHelper
/*
* ip 解析工具類
* */
object IpUtils {
def getCity(ip:String) = {
IpHelper.findRegionByIp(ip)
}
def main(args: Array[String]): Unit = {
println(getCity("58.30.15.255"))
}
}
4.3.5 修改 AccessConvertUtil.scala
val city = IpUtils.getCity(ip)
4.4 資料清洗結果儲存到目標檔案
調優點，控制檔案輸出的大小 coalesce
SparkStatCleanJob.scala
object SparkStatCleanJob {
def main(args: Array[String]): Unit = {
val spark = SparkSession.builder().appName("SparkStatCleanJobApp")
.master("local[2]").getOrCreate()
val accessRDD = spark.sparkContext.textFile("file:///d://access.log")
//accessRDD.take(10).foreach(println)
//RDD -> DF
val accessDF = spark.createDataFrame(accessRDD.map(x => AccessConvertUtil.parseLog(x)),
AccessConvertUtil.struct)
// accessDF.printSchema()
// accessDF.show(false)
accessDF.coalesce(1).write.format("parquet").mode(SaveMode.Overwrite)
.partitionBy("day").save("d://weblog_clean")
spark.stop()
}
}
