title: Spark: grouping of data stream based on cycle time
body: <p>I need your inputs regarding grouping of data stream within spark streaming on the basis of cycle time.</p>
  
  <p>We are receiving input data in this formats  <code>{Object_id:"vm123",  time:"1469077478" , metric :"cpu.usage" , value :"50.8"}</code>.</p>
  
  <p>Data frames are getting ingested very fast at the average rate of 10 seconds.We have a use case to create bins of data based on cycle time .</p>
  
  <p>Suppose Spark bin/batch time is 1 minute for processing of data. Cycle time should be based on the message time-stamp. for example if we receive first packet at 11:30am then we would have to aggregate all messages of that metrics which are received in between 11:30am to 11:31am(1 minute) and send it for processing with cycle time 11.31am.</p>
  
  <p>As per Spark documentation , we only have a support to bin data based on a fix batch duration , for example if we define the batch duration for 1 minute , it will hold the data for 1 minute and send that as a batch , where we have an option to aggregate the data received during this one minuted duration . But this approach does not follows the notion of aggregating the data based on the Cycle Time as defined above.</p>
  
  <p>Please let us know if we have a way to achieve the above use case through Spark or some other tool.</p>
  
  <p>Added Details::
  In Our Use case data frames are getting ingested in each 10 seconds for different entities and each object has few metrics.We need to create bins of data before processing based on cycle time interval (like 5 mins) and the start time of that interval should start with message time-stamp.
  for example:
  We have messages of an object 'vm123' in kafka queue like following:
  message1= {Object_id:"vm123",  time:"t1" , metric :"m1" , value :"50.8"}.
  message2 = {Object_id:"vm123",  time:"t1" , metric :"m2" , value :"55.8"}..............................................................
  Cycle time interval = 5 minutes.
  so a first bin for entity 'VM123' should have all messages of range of  t1 to (t1+5*60)  times and final group of messages with 5 minute cycle time for ob1 should be like following:
  {Object_id:"ob1" , time:"t5" , metrics : [{"name": "m1", value:"average of (v1,v2,v3,v4,v5)},{"name": "m2", value:"average of (v1,v2,v3,v4,v5)}"] }</p>
  
  <p>Thanks </p>