title: Manipulate trigger interval in spark structured streaming
body: <p>For a given scenario, I want to filter the datasets in structured streaming in a combination of continuous and batch triggers. </p>
  
  <p>I know it sounds unrealistic or maybe not feasible. Below is what I am trying to achieve.</p>
  
  <p>Let the processing time-interval set in the app be 5 minutes.
  Let the record be of below schema:</p>
  
  <pre><code>  {
         "type":"record",
         "name":"event",
         "fields":[
           { "name":"Student", "type":"string" },
           { "name":"Subject", "type":"string" } 
     ]}
  </code></pre>
  
  <p>My streaming app is supposed to write the result into the sink by considering either of the below two criteria.</p>
  
  <ol>
  <li><p>If a student is having more than 5 subjects. (Priority to be given to this criteria.)</p></li>
  <li><p>Processing time provided in the trigger expired.</p>
  
  <pre><code>private static Injection&lt;GenericRecord, byte[]&gt; recordInjection;
  private static StructType type;
  public static final String USER_SCHEMA = "{"
          + "\"type\":\"record\","
          + "\"name\":\"alarm\","
          + "\"fields\":["
          + "  { \"name\":\"student\", \"type\":\"string\" },"
          + "  { \"name\":\"subject\", \"type\":\"string\" }"
          + "]}";
  
  private static Schema.Parser parser = new Schema.Parser();
  
  private static Schema schema = parser.parse(USER_SCHEMA);
  
  static {
      recordInjection = GenericAvroCodecs.toBinary(schema);
      type = (StructType) SchemaConverters.toSqlType(schema).dataType();
  
  }
  sparkSession.udf().register("deserialize", (byte[] data) -&gt; {
          GenericRecord record = recordInjection.invert(data).get();
          return RowFactory.create(record.get("student").toString(), record.get("subject").toString());
      }, DataTypes.createStructType(type.fields()));
  
  
  Dataset&lt;Row&gt; ds2 = ds1
          .select("value").as(Encoders.BINARY())
          .selectExpr("deserialize(value) as rows")
          .select("rows.*")
          .selectExpr("student","subject");
  
  StreamingQuery query1 = ds2
          .writeStream()
          .foreachBatch(
              new VoidFunction2&lt;Dataset&lt;Row&gt;, Long&gt;() {
                @Override
                public void call(Dataset&lt;Row&gt; rowDataset, Long aLong) throws Exception {
                  rowDataset.select("student,concat(',',subject)").alias("value").groupBy("student");
                }
              }
          ).format("kafka")
          .option("kafka.bootstrap.servers", "localhost:9092")
          .option("topic", "new_in")
          .option("checkpointLocation", "checkpoint")
          .outputMode("append")
          .trigger(Trigger.ProcessingTime(10000))
          .start();
  query1.awaitTermination();
  </code></pre></li>
  </ol>
  
  <p>Kafka Producer console:</p>
  
  <pre><code>Student:Test, Subject:x
  Student:Test, Subject:y
  Student:Test, Subject:z
  Student:Test1, Subject:x
  Student:Test2, Subject:x
  Student:Test, Subject:w
  Student:Test1, Subject:y
  Student:Test2, Subject:y
  Student:Test, Subject:v
  </code></pre>
  
  <p>In the Kafka consumer console, I am expecting like below.</p>
  
  <pre><code>Test:{x,y,z,w,v} =&gt;This should be the first response 
  Test1:{x,y} =&gt; second
  Test2:{x,y} =&gt; Third by the end of processing time
  </code></pre>