Hi,
I have need to sort the input of the ProcesWindowFunction by one of the
fields of the Tuple4 that is in the Iterator.
static class MyProcessWindowFunction extends
ProcessWindowFunction<Tuple4<Long, Long, String, String>, String, String,
TimeWindow> {
@Override
public void process(String key, Context context,
Iterable<Tuple4<Long, Long, String, String>> input, Collector<String> out)
{
Long elapsed = 0L;
Long pHandlingTime = 0L;
Long totalElapsed = 0L
System.out.println(input.getClass());
Iterator<Tuple4<Long, Long, String, String>> etter =
input.iterator();
*for (Tuple4<Long, Long, String, String> in: input){*
transactionId = in.getField(2).toString();
elapsed = Long.parseLong(in.getField(1).toString()) -
pHandlingTime;
totalElapsed = totalElapsed + elapsed;
pHandlingTime = Long.parseLong(in.getField(1).toString())
out.collect("Key : " + key + " Window : " +
context.window() + " transactionId : " + transactionId + " elapsed : " +
elapsed.toString() + " max handling time : " + h.toString() + "
totalElapsed " + totalElapsed);
}
}
}