日志分析
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)里面的继续切分