Python学习从0到1 day26 第三阶段 Spark ⑤ 搜索引擎日志分析

目录

一、搜索引擎日志分析

二、需求1：热门搜索时间段(小时精度)Top3

实现步骤

三、需求2：打印输出:热门搜索词Top3

实现步骤

四、需求3：打印输出:统计hadoop关键字在哪个时段被搜索最多

实现步骤

五、需求4：将数据转换为JSON格式，写出到文件中

实现步骤

注：

我带着这份勇敢继续向前，忽然明白，我应该是自己的那座山

—— 24.11.10

一、搜索引擎日志分析

原数据文件：（打开百度网盘复制链接）

通过百度网盘分享的文件：search_log.txt
链接：
https://pan.baidu.com/s/1liw33MOGTUn6qdgYFk2SOQ?pwd=1234 
提取码：1234

读取文件转换成RDD，并完成:

① 打印输出:热门搜索时间段(小时精度)Top3

② 打印输出:热门搜索词Top3

③ 打印输出:统计hadoop关键字在哪个时段被搜索最多

④ 将数据转换为JSON格式，写出为文件

二、需求1：热门搜索时间段(小时精度)Top3

实现步骤

① 取出全部的时间并转换为小时

② 转换为(小时，1)的二元元组

③ Key分组聚合Value

④ 排序(降序)

⑤ 取前3

from pyspark import SparkConf, SparkContext
import os
os.environ['PYSPARK_PYTHON'] = "E:/python.learning/pyt/scripts/python.exe"
os.environ['HADOOP_HOME'] = "E:\python.learning\hadoop分布式相关\hadoop-3.0.0"
conf = SparkConf().setMaster("local").setAppName("test_spark")
conf.set("spark.default.parallelize", "1")
sc = SparkContext(conf = conf)

#  读取文件转换成rdd对象
file_rdd = sc.textFile("E:\python.learning\第15章资料\资料\search_log.txt")

# TODO 需求1：热门搜索时间段(小时精度)Top3
# ①取出全部的时间并转换为小时
# ② 转换为(小时，1)的二元元组
# ③ Key分组聚合Value
# ④ 排序(降序)
# ⑤ 取前3
res1 = file_rdd.map(lambda x:x.split("\t")).\
    map(lambda x:x[0][:2]).\
    map(lambda x:(x, 1)).\
    reduceByKey(lambda a,b : a + b).\
    sortBy(lambda x:x[1], ascending = False, numPartitions = 1).\
    take(3)
print(res1)

三、需求2：打印输出:热门搜索词Top3

实现步骤

① 取出全部的搜索词

② (词，1) 二元元组

③ 分组聚合

④ 排序

⑤ 取出Top3

from pyspark import SparkConf, SparkContext
import os
os.environ['PYSPARK_PYTHON'] = "E:/python.learning/pyt/scripts/python.exe"
os.environ['HADOOP_HOME'] = "E:\python.learning\hadoop分布式相关\hadoop-3.0.0"
conf = SparkConf().setMaster("local").setAppName("test_spark")
conf.set("spark.default.parallelize", "1")
sc = SparkContext(conf = conf)

#  读取文件转换成rdd对象
file_rdd = sc.textFile("E:\python.learning\第15章资料\资料\search_log.txt")

# TODO 需求2：打印输出:热门搜索词Top3
# ① 取出全部的搜索词
# ② (词，1) 二元元组
# ③ 分组聚合
# ④ 排序
# ⑤ 取出Top3
file_rdd.map(lambda x : (x.split("\t")[2],1)).\
    reduceBy(lambda a, b : a + b ).\
    sortBy(lambda x : x[1], ascending = False, numPartitions = 1).\
    take(3)

四、需求3：打印输出:统计hadoop关键字在哪个时段被搜索最多

实现步骤

① 过滤内容，只保留hadoop关键词

② 转换为(小时，1)的二元元组

③ Key分组聚合Value

④ 排序(降序)

⑤ 取前1（最多的一个）

from pyspark import SparkConf, SparkContext
import os
os.environ['PYSPARK_PYTHON'] = "E:/python.learning/pyt/scripts/python.exe"
os.environ['HADOOP_HOME'] = "E:\python.learning\hadoop分布式相关\hadoop-3.0.0"
conf = SparkConf().setMaster("local").setAppName("test_spark")
conf.set("spark.default.parallelize", "1")
sc = SparkContext(conf = conf)

#  读取文件转换成rdd对象
file_rdd = sc.textFile("E:\python.learning\第15章资料\资料\search_log.txt")


# TODO 需求3：打印输出:统计hadoop关键字在哪个时段被搜索最多
# ① 过滤内容，只保留hadoop关键词
# ② 转换为(小时，1)的二元元组
# ③ Key分组聚合Value
# ④ 排序(降序)
# ⑤ 取前1（最多的一个）
res3 = file_rdd.map(lambda x : x.split("\t")).\
    filter(lambda x : x[2] == "hadoop").\
    map(lambda x : (x[0][:2] , 1)).\
    reduceByKey(lambda a, b : a + b).\
    sortBy(lambda x : x[1], ascending = False, numPartitions = 1).\
    take(1)
print("res3 : ",res3)

五、需求4：将数据转换为JSON格式，写出到文件中

实现步骤

① 转换为JSON格式的RDD

② 写出为文件

注：

① 每一次链接调用时，都可以在上一层的末尾加上" \ "进行换行，再用” . “进行调用

② 将数据转换为JSON格式最好的方式是先转换为字典，再由字典转换为JSON格式

from pyspark import SparkConf, SparkContext
import os
os.environ['PYSPARK_PYTHON'] = "E:/python.learning/pyt/scripts/python.exe"
os.environ['HADOOP_HOME'] = "E:\python.learning\hadoop分布式相关\hadoop-3.0.0"
conf = SparkConf().setMaster("local").setAppName("test_spark")
conf.set("spark.default.parallelize", "1")
sc = SparkContext(conf = conf)

#  读取文件转换成rdd对象
file_rdd = sc.textFile("E:\python.learning\第15章资料\资料\search_log.txt")

# TODO 需求4：将数据转换为JSON格式，写出到文件中
# ① 转换为JSON格式的RDD
# ② 写出为文件
res4 = file_rdd.map(lambda x : x.split("\t")).\
    map(lambda x : {"time" : x[0], "user_id" : x[1], "key_word" : x[2], "rank1" : x[3], "rank2" : x[4], "url" : x[5]}).\
    saveAsTextFile("E:\python.learning\hadoop分布式相关\data\output4")