Spark数据项目实例二

源数据：https://pan.baidu.com/s/1deTScw6AFnhyHKGbyXiAqg
提取码：ui7c
1.将文件上传至hdfs

hdfs dfs -put /opt/kb09file/op.log /kb09file/

2.导入两个依赖包

import org.apache.spark.sql.types._
import org.apache.spark.sql.functions._

3.读取文件

val fileRDD = sc. textFile("hdfs://hadoop004:9000/kb09file/op.log")

4.查看源文件

fileRDD.collect.foreach(println)

5.将fileRDD进行切割

val jsonStrRDD = fileRDD.map(x => x.split('|')).map(x => (x(0),x(1)))

6.查看jsonStrRDD

jsonStrRDD.collect.foreach(println)

7.取jsonStrRDD的后半部分数据，并将第一部分的数据拼接到后半部分数据的尾部

val jsonRDD =jsonStrRDD.map(x => {var jsonStr = x._2; jsonStr = jsonStr.substring(0,jsonStr.length -1);jsonStr + ",\"id\":" + "\"" + x._1  + "\"" + "}"})

查看jsonRDD

jsonRDD.foreach(println)

8.将jsonRDD转为DataFrame

val jsonDF = jsonRDD.toDF

9.将jsonDF按照cm,ap,et,id字段进行分割

val jsonDF2 = jsonDF.select(get_json_object($"value","$.cm").alias("cm"),get_json_object($"value","$.ap").alias("ap"),get_json_object($"value","$.et").alias("et"),get_json_object($"value","$.id").alias("id"))

10.查看jsonDF2

jsonDF2.show

11.将jsonDF2中的cm段进行分割

val jsonDF3 = jsonDF2.select($"id",$"ap",get_json_object($"cm","$.ln").alias("ln"),get_json_object($"cm","$.sv").alias("sv"),get_json_object($"cm","$.os").alias("os"),get_json_object($"cm","$.g").alias("g"),get_json_object($"cm","$.mid").alias("mid"),get_json_object($"cm","$.l").alias("l"),get_json_object($"cm","$.hw").alias("hw"),get_json_object($"cm","$.uid").alias("uid"),get_json_object($"cm","$.la").alias("la"),get_json_object($"cm","$.vn").alias("vn"),get_json_object($"cm","$.sr").alias("sr"),$"et")

12.查看jsonDF3

jsonDF3.show

13.将jsonDF3进行切割结构化
把字符串

[
{"ett":"a1","en":"a2","kv":"a3"},
{"ett":"a1","en":"a2","kv":"a3"},
{"ett":"a1","en":"a2","kv":"a3"}
]	结构化
ett	en	kv
a1	a2	a3
b1	b2	b3
c1	c2	c3

jsonDF4大脑想到的结构是这样：
id ap mid nw … event

val jsonDF4 = jsonDF3.select($"id",$"ap",$"ln",$"sv",$"os",$"g",$"mid",$"nw",$"l",$"vc",$"hw",$"ar",$"uid",$"t",$"la",$"md",$"vn",$"ba",$"sr",from_json($"et",ArrayType(StructType(StructField("ett",StringType)::StructField("en",StringType)::StructField("kv",StringType)::Nil))).alias("event"))

14.将jsonDF4中的event段展开

val jsonDF5 = jsonDF4.select($"id",$"ap",$"ln",$"sv",$"os",$"g",$"mid",$"nw",$"l",$"vc",$"hw",$"ar",$"uid",$"t",$"la",$"md",$"vn",$"ba",$"sr",explode($"event").alias("event")).show

15.将jsonDF5中的ett、en、kv展开

val jsonDF6 = jsonDF5.select($"id",$"ap",$"ln",$"sv",$"os",$"g",$"mid",$"nw",$"l",$"vc",$"hw",$"ar",$"uid",$"t",$"la",$"md",$"vn",$"ba",$"sr",$"event.ett",$"event.en",$"event.kv")

16.将en中的loading过滤出来

 val loadingDF1 = jsonDF6.filter("en = 'loading'")

17.将loading中的数据按照
“extend2”: “”,
“loading_time”: “0”,
“action”: “3”,
“extend1”: “”,
“type”: “1”,
“type1”: “102”,
“loading_way”: "1"进行切割

 val loadingDF2 = loadingDF1.select($"id",$"ap",$"ln",$"sv",$"os",$"g",$"mid",$"nw",$"l",$"vc",$"hw",$"ar",$"uid",$"t",$"la",$"md",$"vn",$"ba",$"sr",$"ett",$"en",get_json_object($"kv","$.extend2").alias("extend2"),get_json_object($"kv","$.loading_time").alias("loading_time"),get_json_object($"kv","$.action").alias("action"),get_json_object($"kv","$.extend1").alias("extend1"),get_json_object($"kv","$.type").alias("type"),get_json_object($"kv","$.type1").alias("type1"),get_json_object($"kv","$.loading_way").alias("loading_way"))

18.将en中的ad过滤出来

 val adDF1 = jsonDF6.filter("en = 'ad'")

19.将ad进行格式化，字段如下
“activityId”:“1”,
“displayMills”:“78522”,
“entry”:“1”,
“action”:“1”,
“contentType”:“0”

 val adDF2 = adDF1.select($"id",$"ap",$"ln",$"sv",$"os",$"g",$"mid",$"nw",$"l",$"vc",$"hw",$"ar",$"uid",$"t",$"la",$"md",$"vn",$"ba",$"sr",$"ett",$"en",get_json_object($"kv","$.activityId").alias("activityId"),get_json_object($"kv","$.displayMills").alias("displayMills"),get_json_object($"kv","$.entry").alias("entry"),get_json_object($"kv","$.action").alias("action"),get_json_object($"kv","$.contentType").alias("contentType"))

20.将en中的notification过滤出来

val notificationDF1 = jsonDF6.filter("en = 'notification'")

21.将notification进行格式化，字段如下
“ap_time”:“1593097087883”,
“action”:“1”,
“type”:“1”,
“content”:""

val notificationDF2 = notificationDF1.select($"id",$"ap",$"ln",$"sv",$"os",$"g",$"mid",$"nw",$"l",$"vc",$"hw",$"ar",$"uid",$"t",$"la",$"md",$"vn",$"ba",$"sr",$"ett",$"en",get_json_object($"kv","$.ap_time").alias("ap_time"),get_json_object($"kv","$.action").alias("action"),get_json_object($"kv","$.type").alias("type"),get_json_object($"kv","$.content").alias("content"))

22.将en中的coment过滤出来

 val commentDF1 = jsonDF6.filter("en = 'comment'")

23.将coment进行格式化，字段如下
“p_comment_id”:1,
“addtime”:“1593097573725”,
“praise_count”:973,
“other_id”:5,
“comment_id”:9,
“reply_count”:40,
“userid”:7,
“content”:“辑赤蹲慰鸽抿肘捎”

val commentDF2 = commentDF1.select($"id",$"ap",$"ln",$"sv",$"os",$"g",$"mid",$"nw",$"l",$"vc",$"hw",$"ar",$"uid",$"t",$"la",$"md",$"vn",$"ba",$"sr",$"ett",$"en",get_json_object($"kv","$.p_comment_id").alias("p_comment_id"),get_json_object($"kv","$.addtime").alias("addtime"),get_json_object($"kv","$.praise_count").alias("praise_count"),get_json_object($"kv","$.other_id").alias("other_id"),get_json_object($"kv","$.comment_id").alias("comment_id"),get_json_object($"kv","$.reply_count").alias("reply_count"),get_json_object($"kv","$.userid").alias("userid"),get_json_object($"kv","$.content").alias("content"))

24.将en中的active_background过滤出来

val active_backgroundDF1 = jsonDF6.filter("en = 'active_background'")

25.将active_background格式化，字段如下
“active_source”:“3”

val active_backgroundDF2 = active_backgroundDF1.select($"id",$"ap",$"ln",$"sv",$"os",$"g",$"mid",$"nw",$"l",$"vc",$"hw",$"ar",$"uid",$"t",$"la",$"md",$"vn",$"ba",$"sr",$"ett",$"en",get_json_object($"kv","$.active_source").alias("active_source"))

26.将en中的praise过滤出来

 val praiseDF1 = jsonDF6.filter("en = 'praise'")

27.将praise格式化，字段如下
“target_id”: 9,
“id”: 7,
“type”: 1,
“add_time”: “1593098545976”,
“userid”: 8

val praiseDF2 = praiseDF1.select($"id",$"ap",$"ln",$"sv",$"os",$"g",$"mid",$"nw",$"l",$"vc",$"hw",$"ar",$"uid",$"t",$"la",$"md",$"vn",$"ba",$"sr",$"ett",$"en",get_json_object($"kv","$.target_id").alias("target_id"),get_json_object($"kv","$.id").alias("id"),get_json_object($"kv","$.type").alias("type"),get_json_object($"kv","$.add_time").alias("add_time"),get_json_object($"kv","$.userid").alias("userid"))