We need create a stream by csql() and then use this stream as the other stream's input. I found if I use returns() to create the first stream the application work very well, but if I use returnAsRow() or returnAsMap() to create the first stream I will get the ArrayIndexOutofBounds exception.
But we know that the returns() function use tuple and the tuple has a limitation of 25. So I create a function named returnsTransformRow() to accompolish my goal by row.
We need create a stream by csql() and then use this stream as the other stream's input. I found if I use returns() to create the first stream the application work very well, but if I use returnAsRow() or returnAsMap() to create the first stream I will get the ArrayIndexOutofBounds exception. But we know that the returns() function use tuple and the tuple has a limitation of 25. So I create a function named returnsTransformRow() to accompolish my goal by row.