基于Spark的大規模日志分析-有解無憂

摘要：本篇文章將從一個實際專案出發，分享如何使用 Spark 進行大規模日志分析，并通過代碼演示加深讀者的理解，

本文分享自華為云社區《【實戰經驗分享】基于Spark的大規模日志分析【上進小菜豬大資料系列】》，作者：上進小菜豬，

隨著互聯網的普及和應用范圍的擴大，越來越多的應用場景需要對海量資料進行高效地處理和分析，這就要求我們必須具備大資料技術方面的知識和技能，本篇文章將從一個實際專案出發，分享如何使用 Spark 進行大規模日志分析，并通過代碼演示加深讀者的理解，

1.資料來源

我們的專案是針對某購物網站的訪問日志進行分析，其中主要包含以下幾個欄位：

IP：訪問的客戶端 IP 地址
Time：訪問時間
Url：訪問的 URL 地址
User-Agent：瀏覽器識別符號

原始資料規模約為 100GB，我們需要對其進行清洗、統計和分析，以得到有用的資訊和價值，

2. 資料清洗

由于原始資料存在缺失值、例外值、重復值等問題，因此我們需要進行資料清洗，主要包括以下步驟：

將原始資料進行格式轉換，方便后續處理
對 IP、Time、Url 和 User-Agent 欄位進行決議和提取
去除不合法的記錄和重復的記錄

具體代碼實作如下：

import org.apache.spark.{SparkConf, SparkContext}
import java.text.SimpleDateFormat
import java.util.Locale
?
object DataCleaning {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("DataCleaning")
    val sc = new SparkContext(conf)
    val data = sc.textFile("hdfs://master:9000/log/access.log")
?
 // 定義時間格式及地區資訊
    val dateFormat = new SimpleDateFormat("dd/MMM/yyyy:HH:mm:ss Z", Locale.ENGLISH)
?
 // 資料清洗
    val cleanData = https://www.cnblogs.com/huaweiyun/p/data.map(line => {
      val arr = line.split(" ")
 if (arr.length >= 9) {
 // 決議 IP
        val ip = arr(0)
?
 // 決議時間，轉換為 Unix 時間戳
        val time = dateFormat.parse(arr(3) + " " + arr(4)).getTime / 1000
?
 // 決議 URL
        val url = urlDecode(arr(6))
?
 // 決議 UserAgent
        val ua = arr(8)
?
 (ip, time, url, ua)
 }
 }).filter(x => x != null).distinct()
?
 // 結果輸出
    cleanData.saveAsTextFile("hdfs://master:9000/cleanData")
?
    sc.stop()
 }
?
 // URL 解碼
  def urlDecode(url: String): String = {
    java.net.URLDecoder.decode(url, "utf-8")
 }
}

3. 資料統計

對于大規模資料的處理，我們可以使用 Spark 提供的強大的分布式計算能力，以提高處理效率和減少計算時間，

我們這里使用 Spark SQL 統計每個 URL 的訪問量，并輸出前 10 個訪問量最高的 URL，代碼如下：

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SQLContext
?
case class LogRecord(ip: String, time: Long, url: String, ua: String)
?
object DataAnalysis {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("DataAnalysis")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
?
 // 讀取清洗后的資料
    val cleanData = https://www.cnblogs.com/huaweiyun/p/sc.textFile("hdfs://master:9000/cleanData").filter(x => x != null)
?
 // 將資料轉換為 DataFrame
 import sqlContext.implicits._
    val logDF = cleanData.map(_.split(",")).map(p => LogRecord(p(0), p(1).toLong, p(2), p(3))).toDF()
?
 // 統計每個 URL 的訪問量，并按訪問量降序排序
    val topUrls = logDF.groupBy("url").count().sort($"count".desc)
?
 // 輸出前 10 個訪問量最高的 URL
    topUrls.take(10).foreach(println)
?
    sc.stop()
 }
}

4. 資料可視化

資料可視化是將處理和分析后的資料以圖表或影像的方式展示出來，有利于我們直觀地觀察資料的規律和趨勢，

我們這里采用 Python 的 Matplotlib 庫將前 10 個訪問量最高的 URL 可視化，代碼如下：

import matplotlib.pyplot as plt
?
# 讀取資料
with open('topUrls.txt', 'r') as f:
    line = f.readline()
    urls = []
    counts = []
 while line and len(urls) < 10:
        url, count = line.strip().split(',')
        urls.append(url)
        counts.append(int(count))
        line = f.readline()
# 繪制直方圖
plt.bar(range(10), counts, align='center')
plt.xticks(range(10), urls, rotation=90)
plt.xlabel('Url')
plt.ylabel('Count')
plt.title('Top 10 Url')
plt.show()

在進行資料清洗前，需要先對原始日志資料進行篩選，選取需要分析的欄位，然后進行資料清洗，去掉不必要的空格、特殊字符等，使資料更加規整，并增加可讀性，

下面是資料清洗的代碼示例：

val originalRdd = spark.sparkContext.textFile("path/to/logfile")
?
val filteredRdd = originalRdd.filter(line => {
  val tokens = line.split("\t")
  tokens.length >= 10 &&
 tokens(0).matches("\d{4}-\d{2}-\d{2}") &&
 tokens(1).matches("\d{2}:\d{2}:\d{2}") &&
 tokens(2).matches("\d+") &&
 tokens(3).matches("\d+") &&
 tokens(4).matches("\d+") &&
 tokens(5).matches("\d+") &&
 tokens(6).matches(".+") &&
 tokens(7).matches(".+") &&
 tokens(8).matches(".+") &&
 tokens(9).matches(".+")
})
?
val cleanedRdd = filteredRdd.map(line => {
  val tokens = line.split("\t")
  val timestamp = s"${tokens(0)} ${tokens(1)}"
  val request = tokens(6).replaceAll(""", "")
  val responseCode = tokens(8).toInt
 (timestamp, request, responseCode)
})

?在上述代碼中，我們首先讀取原始日志資料，并使用filter函式過濾掉不符合條件的行；然后使用map函式將資料轉換為元組的形式，并進行清洗，其中，元組的三個元素分別是時間戳、請求內容和回應狀態碼，

接下來，讓我們來介紹一下如何使用Spark進行資料統計，

資料統計是大規模資料分析中非常重要的一個環節，Spark提供了豐富的聚合函式，可用于對資料進行各種統計分析，

下面是對清洗后的資料進行統計分析的代碼示例：

import org.apache.spark.sql.functions._
?
val df = spark.createDataFrame(cleanedRdd).toDF("timestamp", "request", "responseCode")
val totalCount = df.count()
val errorsCount = df.filter(col("responseCode") >= 400).count()
val successCount = totalCount - errorsCount
val topEndpoints = df.groupBy("request").count().orderBy(desc("count")).limit(10)
topEndpoints.show()

在上面的代碼中，我們首先將清洗后的資料轉換為DataFrame，然后使用count函式計算總記錄數和錯誤記錄數，并計算成功記錄數，最后使用groupBy和orderBy函式按照請求內容，對資料進行分組統計，并列印出請求次數最多的前10個端點，

通過可視化，我們可以清楚地看到前 10 個訪問量最高的 URL 地址及其訪問量，這對于進一步分析和優化網站的性能和用戶體驗具有重要的意義，

總結起來，這就是我們的一個大資料實戰專案，我們使用 Spark 統計了購物網站的訪問量，并通過 Python 的 Matplotlib 庫將結果可視化，這個程序中，我們運用了資料清洗、Spark SQL 統計和可視化等技術，為大規模資料的處理和分析提供了有效的解決方案，

點擊關注，第一時間了解華為云新鮮技術~

轉載請註明出處，本文鏈接：https://www.uj5u.com/qiye/555285.html

標籤：JavaScript

上一篇：利用jQuery制作簡易的table上下無縫輪播

下一篇：返回列表