spark SQL学习（综合案例-日志分析）
2024-09-01 04:53:24
日志分析
scala> import org.apache.spark.sql.types._
scala> import org.apache.spark.sql.Row
scala> val logRDD = sc.textFile("hdfs://master:9000/student/2016113012/data/log.txt").map(_.split("#"))
logRDD: org.apache.spark.rdd.RDD[Array[String]] = MapPartitionsRDD[2] at map at <console>:21
val schema = StructType(
Array(
StructField("ipAddress",StringType,true),
StructField("clientIndentd",StringType,true),
StructField("userId",StringType,true),
StructField("dateTime",StringType,true),
StructField("protocal",StringType,true),
StructField("responseCode",StringType,true),
StructField("contentSize",IntegerType,true)
)
)
val rowRDD = logRDD.map(p => Row(p(0),p(1),p(2),p(3),p(4),p(5),p(6).toInt))
val logDF = sqlContext.createDataFrame(rowRDD,schema)
logDF.registerTempTable("logs")
//统计访问文件大小的平均值，最大值，最小值
scala> sqlContext.sql("select avg(contentSize),min(contentSize),max(contentSize) from logs").show()
17/03/07 17:04:20 INFO ParseDriver: Parsing command: select avg(contentSize),min(contentSize),max(contentSize) from logs
17/03/07 17:04:20 INFO ParseDriver: Parse Completed
17/03/07 17:04:21 INFO FileInputFormat: Total input paths to process : 1
17/03/07 17:04:22 INFO deprecation: mapred.tip.id is deprecated. Instead, use mapreduce.task.id
17/03/07 17:04:22 INFO deprecation: mapred.task.id is deprecated. Instead, use mapreduce.task.attempt.id
17/03/07 17:04:22 INFO deprecation: mapred.task.is.map is deprecated. Instead, use mapreduce.task.ismap
17/03/07 17:04:22 INFO deprecation: mapred.task.partition is deprecated. Instead, use mapreduce.task.partition
17/03/07 17:04:22 INFO deprecation: mapred.job.id is deprecated. Instead, use mapreduce.job.id
+------+----+----+
| _c0| _c1| _c2|
+------+----+----+
|3506.0|2000|5554|
+------+----+----+
//统计响应代码的数量
scala> sqlContext.sql("select responseCode,count(*) from logs group by responseCode").show()
17/03/07 17:52:26 INFO ParseDriver: Parsing command: select responseCode,count(*) from logs group by responseCode
17/03/07 17:52:26 INFO ParseDriver: Parse Completed
+------------+---+
|responseCode|_c1|
+------------+---+
| 304| 1|
| 200| 2|
+------------+---+
//统计大于1次的ip地址
scala> sqlContext.sql("select ipAddress,count(1) as total from logs group by ipAddress having total > 1").show()
17/03/07 17:55:20 INFO ParseDriver: Parsing command: select ipAddress,count(1) as total from logs group by ipAddress having total > 1
17/03/07 17:55:20 INFO ParseDriver: Parse Completed
+----------+-----+
| ipAddress|total|
+----------+-----+
|10.0.0.153| 3|
+----------+-----+
问题：如何将p(4)里面的继续切分
最新文章
- Spring 4 异常处理
- OO.A.D.P
- 深入理解计算机系统（2.8）---浮点数的舍入，Java中的舍入例子以及浮点数运算（重要）
- 如何在 Linux 终端中知道你的公有 IP
- Eclipse开发C++遇到的Binary not found等问题
- php中读取文件内容的几种方法
- Jenkins入门-转
- maven添加oracle jdbc依赖
- linux_coom _ Linux文件比较，文本文件的交集、差集与求差
- zf-关于公司框架的时间字段的格式转换问题。。
- jQuery.extend 使用函数
- 第九条：覆盖equals方法时总要覆盖hashCode方法
- JQuery实战---初识JQuery+入门实例
- JS&Java实现常见算法面试题
- ios打包unity应用以及配置签名
- Centos7.3下安装Jumpserver 1.0.0（支持windows组件）
- IOS渗透测试第一步-基础知识统一放送
- 在delphi XE5 里面编译kbmmw4.3
- python编程之socket编程基础
- Hibernate- 动态实例查询
热门文章
- docker remote api enable in ubuntu
- Logstash之时区问题的建议和修改---filter---and duplicate resolution.
- 属性attribute和property的区别
- 如何使用 libtorch 实现 VGG16 网络？
- 使用Standford coreNLP进行中文命名实体识别
- Android集成百度地图SDK
- Java设计模式-工厂方法模式（Virtual Constructor/Polymorphic Factory）
- testng的@Factory及其与@DataProvider的区别
- request.getInputStream() 流只能读取一次问题
- 自定义admin管理工具(stark组件)