Spark學習筆記（6）—— 網站訪問次數統計
阿新 • • 發佈：2018-12-16
1 資料檔案
20160321101954 http://java.itcast.cn/java/course/javaeeadvanced.shtml 20160321101954 http://java.itcast.cn/java/course/javaee.shtml 20160321101954 http://java.itcast.cn/java/course/android.shtml 20160321101954 http://java.itcast.cn/java/video.shtml 20160321101954 http://java.itcast.cn/java/teacher.shtml 20160321101954 http://java.itcast.cn/java/course/android.shtml 20160321101954 http://php.itcast.cn/php/teacher.shtml 20160321101954 http://net.itcast.cn/net/teacher.shtml 20160321101954 http://java.itcast.cn/java/course/hadoop.shtml 20160321101954 http://java.itcast.cn/java/course/base.shtml 20160321101954 http://net.itcast.cn/net/course.shtml 20160321101954 http://php.itcast.cn/php/teacher.shtml 20160321101954 http://net.itcast.cn/net/video.shtml 20160321101954 http://java.itcast.cn/java/course/base.shtml 20160321101954 http://net.itcast.cn/net/teacher.shtml 20160321101954 http://java.itcast.cn/java/video.shtml 20160321101954 http://java.itcast.cn/java/video.shtml 20160321101954 http://net.itcast.cn/net/video.shtml 20160321101954 http://net.itcast.cn/net/course.shtml 20160321101954 http://java.itcast.cn/java/course/javaee.shtml 20160321101954 http://java.itcast.cn/java/course/android.shtml ..........
2 原始碼
2.1 測試1
package webcount
import org.apache.spark.{SparkConf, SparkContext}
object UrlCount {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setAppName("URLCount").setMaster("local[2]")
val sc = new SparkContext(conf)
//rdd1 將資料切分，然後元組中（url,1）
val rdd1 = sc. ArrayBuffer((http://php.itcast.cn/php/course.shtml,459), (http://java.itcast.cn/java/course/base.shtml,543), (http://java.itcast.cn/java/video.shtml,496), (http://java.itcast.cn/java/course/android.shtml,501), (http://net.itcast.cn/net/video.shtml,521), (http://java.itcast.cn/java/course/hadoop.shtml,506), (http://net.itcast.cn/net/course.shtml,521), (http://java.itcast.cn/java/course/cloud.shtml,1028), (http://php.itcast.cn/php/video.shtml,490), (http://java.itcast.cn/java/teacher.shtml,482), (http://php.itcast.cn/php/teacher.shtml,464), (http://net.itcast.cn/net/teacher.shtml,512), (http://java.itcast.cn/java/course/javaee.shtml,1000), (http://java.itcast.cn/java/course/javaeeadvanced.shtml,477))
2.2 測試2
package webcount
import java.net.URL
import org.apache.spark.{SparkConf, SparkContext}
object UrlCount {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setAppName("URLCount").setMaster("local[2]")
val sc = new SparkContext(conf)
//rdd1 將資料切分，然後元組中（url,1）
val rdd1 = sc.textFile("d://urlcount.log").map(line => {
val f = line.split("\t")
(f(1), 1)
})
val rdd2 = rdd1.reduceByKey(_ + _)
val rdd3 = rdd2.map(t => {
val url = t._1
val host = new URL(url).getHost
(host, url, t._2)
})
val rdd4 = rdd3.groupBy(_._1)
println(rdd4.collect().toBuffer)
sc.stop()
}
}
ArrayBuffer((net.itcast.cn,CompactBuffer((net.itcast.cn,http://net.itcast.cn/net/video.shtml,521), (net.itcast.cn,http://net.itcast.cn/net/course.shtml,521),
(net.itcast.cn,http://net.itcast.cn/net/teacher.shtml,512))), (java.itcast.cn,CompactBuffer((java.itcast.cn,http://java.itcast.cn/java/course/base.shtml,543),
(java.itcast.cn,http://java.itcast.cn/java/video.shtml,496), (java.itcast.cn,http://java.itcast.cn/java/course/android.shtml,501),
(java.itcast.cn,http://java.itcast.cn/java/course/hadoop.shtml,506), (java.itcast.cn,http://java.itcast.cn/java/course/cloud.shtml,1028),
(java.itcast.cn,http://java.itcast.cn/java/teacher.shtml,482), (java.itcast.cn,http://java.itcast.cn/java/course/javaee.shtml,1000),
(java.itcast.cn,http://java.itcast.cn/java/course/javaeeadvanced.shtml,477))), (php.itcast.cn,CompactBuffer((php.itcast.cn,http://php.itcast.cn/php/course.shtml,459),
(php.itcast.cn,http://php.itcast.cn/php/video.shtml,490), (php.itcast.cn,http://php.itcast.cn/php/teacher.shtml,464))))
2.3 測試3
package webcount
import java.net.URL
import org.apache.spark.{SparkConf, SparkContext}
object UrlCount {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setAppName("URLCount").setMaster("local[2]")
val sc = new SparkContext(conf)
//rdd1 將資料切分，然後元組中（url,1）
val rdd1 = sc.textFile("d://urlcount.log").map(line => {
val f = line.split("\t")
(f(1), 1)
})
val rdd2 = rdd1.reduceByKey(_ + _)
val rdd3 = rdd2.map(t => {
val url = t._1
val host = new URL(url).getHost
(host, url, t._2)
})
val rdd4 = rdd3.groupBy(_._1).mapValues(it =>{
it.toList.sortBy(_._3).reverse.take(3)
})
println(rdd4.collect().toBuffer)
sc.stop()
}
}
ArrayBuffer((net.itcast.cn,List((net.itcast.cn,http://net.itcast.cn/net/course.shtml,521), (net.itcast.cn,http://net.itcast.cn/net/video.shtml,521),
(net.itcast.cn,http://net.itcast.cn/net/teacher.shtml,512))), (java.itcast.cn,List((java.itcast.cn,http://java.itcast.cn/java/course/cloud.shtml,1028),
(java.itcast.cn,http://java.itcast.cn/java/course/javaee.shtml,1000), (java.itcast.cn,http://java.itcast.cn/java/course/base.shtml,543))),
(php.itcast.cn,List((php.itcast.cn,http://php.itcast.cn/php/video.shtml,490), (php.itcast.cn,http://php.itcast.cn/php/teacher.shtml,464),
(php.itcast.cn,http://php.itcast.cn/php/course.shtml,459))))
3 改進版
上個版本採用list 存資料，如果資料太多就會奔潰，改用RDD,當資料量大時，它會存到磁碟上。
3.1 測試1
package webcount
import java.net.URL
import org.apache.spark.{SparkConf, SparkContext}
object AdvUrlCount {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setAppName("AdvURLCount").setMaster("local[2]")
val sc = new SparkContext(conf)
//rdd1 將資料切分，然後元組中（url,1）
val rdd1 = sc.textFile("d://urlcount.log").map(line => {
val f = line.split("\t")
(f(1), 1)
})
val rdd2 = rdd1.reduceByKey(_ + _)
val rdd3 = rdd2.map(t => {
val url = t._1
val host = new URL(url).getHost
(host, url, t._2)
})
println(rdd3.collect().toBuffer)
// val rddjava = rdd3.filter(_._1 == "java.itcast.cn")
sc.stop()
}
}
ArrayBuffer((php.itcast.cn,http://php.itcast.cn/php/course.shtml,459), (java.itcast.cn,http://java.itcast.cn/java/course/base.shtml,543),
(java.itcast.cn,http://java.itcast.cn/java/video.shtml,496), (java.itcast.cn,http://java.itcast.cn/java/course/android.shtml,501),
(net.itcast.cn,http://net.itcast.cn/net/video.shtml,521), (java.itcast.cn,http://java.itcast.cn/java/course/hadoop.shtml,506),
(net.itcast.cn,http://net.itcast.cn/net/course.shtml,521), (java.itcast.cn,http://java.itcast.cn/java/course/cloud.shtml,1028),
(php.itcast.cn,http://php.itcast.cn/php/video.shtml,490), (java.itcast.cn,http://java.itcast.cn/java/teacher.shtml,482),
(php.itcast.cn,http://php.itcast.cn/php/teacher.shtml,464), (net.itcast.cn,http://net.itcast.cn/net/teacher.shtml,512),
(java.itcast.cn,http://java.itcast.cn/java/course/javaee.shtml,1000), (java.itcast.cn,http://java.itcast.cn/java/course/javaeeadvanced.shtml,477))
3.2 測試2
....
val rddjava = rdd3.filter(_._1 == "java.itcast.cn")
println(rddjava.collect().toBuffer)
ArrayBuffer((java.itcast.cn,http://java.itcast.cn/java/course/base.shtml,543), (java.itcast.cn,http://java.itcast.cn/java/video.shtml,496),
(java.itcast.cn,http://java.itcast.cn/java/course/android.shtml,501), (java.itcast.cn,http://java.itcast.cn/java/course/hadoop.shtml,506),
(java.itcast.cn,http://java.itcast.cn/java/course/cloud.shtml,1028), (java.itcast.cn,http://java.itcast.cn/java/teacher.shtml,482),
(java.itcast.cn,http://java.itcast.cn/java/course/javaee.shtml,1000), (java.itcast.cn,http://java.itcast.cn/java/course/javaeeadvanced.shtml,477))
3.3 測試3
val rddjava = rdd3.filter(_._1 == "java.itcast.cn")
val sortedjava = rddjava.sortBy(_._3, false).take(3)
println(sortedjava.toBuffer)
ArrayBuffer((java.itcast.cn,http://java.itcast.cn/java/course/cloud.shtml,1028),
(java.itcast.cn,http://java.itcast.cn/java/course/javaee.shtml,1000),
(java.itcast.cn,http://java.itcast.cn/java/course/base.shtml,543))
3.4 測試4
package webcount
import java.net.URL
import org.apache.spark.{SparkConf, SparkContext}
object AdvUrlCount {
def main(args: Array[String]): Unit = {
//從資料庫中載入規則
val arr = Array("java.itcast.cn", "php.itcast.cn", "net.itcast.cn")
val conf = new SparkConf().setAppName("AdvURLCount").setMaster("local[2]")
val sc = new SparkContext(conf)
//rdd1 將資料切分，然後元組中（url,1）
val rdd1 = sc.textFile("d://urlcount.log").map(line => {
val f = line.split("\t")
(f(1), 1)
})
val rdd2 = rdd1.reduceByKey(_ + _)
val rdd3 = rdd2.map(t => {
val url = t._1
val host = new URL(url).getHost
(host, url, t._2)
})
//println(rdd3.collect().toBuffer)
for (institute <- arr) {
val rdd = rdd3.filter(_._1 == institute)
val result = rdd.sortBy(_._3, false).take(3)
/*
* 通過JDBC 向資料庫儲存資料
* id:學院，URL,次數，訪問日期
* */
println(result.toBuffer)
}
sc.stop()
}
}
ArrayBuffer((java.itcast.cn,http://java.itcast.cn/java/course/cloud.shtml,1028),
(java.itcast.cn,http://java.itcast.cn/java/course/javaee.shtml,1000),
(java.itcast.cn,http://java.itcast.cn/java/course/base.shtml,543))
ArrayBuffer((php.itcast.cn,http://php.itcast.cn/php/video.shtml,490),
(php.itcast.cn,http://php.itcast.cn/php/teacher.shtml,464),
(php.itcast.cn,http://php.itcast.cn/php/course.shtml,459))
ArrayBuffer((net.itcast.cn,http://net.itcast.cn/net/video.shtml,521),
(net.itcast.cn,http://net.itcast.cn/net/course.shtml,521),
(net.itcast.cn,http://net.itcast.cn/net/teacher.shtml,512))