pv、uv、ipを5分ごとに分析するスパークを使用する方法

Question

5分ごとにuv、pv、ipを分析し、Mysqlを格納する方法。

Message sent: {"cookie":"a95f22eabc4fd4b580c011a3161a9d9d","ip":"125.119.144.252","event_time":"2017-08-07 10:50:16"} 
Message sent: {"cookie":"6b67c8c700427dee7552f81f3228c927","ip":"202.109.201.181","event_time":"2017-08-07 10:50:26"} 

それはちょうど00のようなものです：：00-00：05 00：05--00：10というように、 私が使用：データは次の形式でカフカからである

val write=new JDBCSink() 
     val query=counts.writeStream.foreach(write).outputMode("complete") 
      .trigger(ProcessingTime("5 minutes"))  
      .start() 

が、私が00:01にコミットするか、それが故障すると、どうすれば00：01-00：06のように分析されないのでしょうか。

Answer 1

window機能の使用：

query = counts.groupBy(window('event_time', '5 second')).agg() 
query.writeStream.start()