spark SQL學習(綜合案例-日誌分析)
阿新 • • 發佈:2019-01-10
日誌分析
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)裡面的繼續切分