title: Spark Streaming with mapGroupsWithState
body: <p>I am writing a Stateful Streaming application in which I am using mapGroupsWithState to create aggregates for Groups but I need to create <strong>Groups based on more than one column in the Input Row</strong>. All the examples in the 'Spark: The Definitive Guide' use only one column such as 'User' or 'Device'. I am using code similar to what's given below. <strong>How do I specify more than one field in the 'groupByKey'?</strong></p>
  
  <p>There are other challenges as well. The book says we can use 'updateAcrossEvents' the way given below but I get compile time error saying: Error:(43, 65) <strong>missing argument list for method updateAcrossEvents</strong> in object Main
  Unapplied methods are only converted to functions when a function type is expected.
  You can make this conversion explicit by writing <code>updateAcrossEvents _</code> or <code>updateAcrossEvents(_,_,_,_,_)</code> instead of <code>updateAcrossEvents</code>.
        .mapGroupsWithState(GroupStateTimeout.EventTimeTimeout())(updateAcrossEvents)</p>
  
  <p>Another challenge: Compiler also complains about the my MyReport: Error:(41, 12) Unable to find encoder for type stored in a Dataset.  Primitive types (Int, String, etc) and Product types (case classes) are supported by importing spark.implicits._  <strong>Support for serializing other types will be added in future releases.</strong>
          </p>
  
  <p>Help in resolving these errors would be greatly appreciated. Thanks in advance.</p>
  
  <pre><code>withEventTime
   .as[MyReport]
   .groupByKey(_.getKeys.getKey1). // How do I add _.getKeys.getKey2?
   .mapGroupsWithState(GroupStateTimeout.EventTimeTimeout())(updateAcrossEvents)
   .writeStream
   .queryName("test_query")
   .format("memory")
   .outputMode("update")
   .start()
  </code></pre>
  
  <p><strong>updateAcrossEvents:</strong></p>
  
  <pre><code>def updateAcrossEvents (tuple3: Tuple3[String, String, String], inputs: Iterator[MyReport], oldState: GroupState[MyState]): MyState = {
  
   var state: MyState = if (oldState.exists) oldState.get else MyState.getNewState(tuple3._1, tuple3._2, tuple3._3)
  
   for (input &lt;- inputs) {
   state = updateWithEvent(state, input)
   oldState.update(state)
   }
  
   state
  }
  </code></pre>
  
  <p><strong>updateWithEvent:</strong></p>
  
  <pre><code>def updateWithEvent(state: MyState, report: MyReport): MyState = {
  
   state.someField1 = state.someField1 ++ Array(report.getSomeField1.longValue())
   state.someField2 = state.someField2 ++ Array(report.getSomeField2.longValue())
  
   state
  }
  </code></pre>