You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by nsengupta <se...@gmail.com> on 2017/02/14 19:18:41 UTC
Clarification: use of AllWindowedStream.apply() function
I am trying to understand if the AllWindowedStream.apply() function can be
used for creating a DataStream of new types.
Here is a portion of the code:
------------------------------------------------------------------------------------------------------------------------
case class RawMITSIMTuple(
tupletype: Int, timeOfReport: Int,
vehicleID: Int, vehicleSpeed: Int,
expressWayID: Int, vehicleLane: Int,
vehicleDir: Int,
vehicleSegment: Int, vehiclePos: Int, queyID:
Int,
segmentInit: Int, segmentEnd: Int ,
dayOfWeek: Int, timeOfDay: Int, dayID:
Int
)
case class EWayCoordinates(eWayID: Int, eWayDir: Int, eWayLane: Int,
eWaySegment: Int)
case class VehicleDetails(vehicleID: Int, vehicleSpeed: Int, vehiclePos:
Int)
case class PositionReport(
tupletype: Int, timeOfReport: Int,
eWayCoordinates: EWayCoordinates,
vehicleDetails: VehicleDetails
)
val envDefault = StreamExecutionEnvironment.getExecutionEnvironment
envDefault.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
// ...
val positionReportStream = this
.readRawMITSIMTuplesInjected(envDefault,args(0))
.assignAscendingTimestamps(e => e.timeOfReport)
.windowAll(TumblingEventTimeWindows.of(Time.seconds(30)))
------------------------------------------------------------------------------------------------------------------------
positionReportStream above is of type *AllWindowedStream*. As such, I cannot
use it as a DataStream[PositionReport]: I cannot segregate it by some kind
of KeySelection and use it further down.
I have been thinking of using a FoldFunction on it, but that gives a
collection of PositionReport. So, I get a DataStream[Vector[PositionReport]]
(Vector is just an example).
The other alternative is to use an AllWindowedStream.apply() function, where
I can emit a DataStream[PositionReport]. But, that will mean that I am using
the apply function more as a *mapper*. Is that the right way to use it?
Could someone please push me to the correct way to deal with it?
-- Nirmalya
--
View this message in context: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Clarification-use-of-AllWindowedStream-apply-function-tp11627.html
Sent from the Apache Flink User Mailing List archive. mailing list archive at Nabble.com.
Re: Clarification: use of AllWindowedStream.apply() function
Posted by Aljoscha Krettek <al...@apache.org>.
Yes, you're correct. :-)
On Thu, 16 Feb 2017 at 14:24 nsengupta <se...@gmail.com> wrote:
> Thanks, Aljoscha for the clarification.
>
> I understand that instead of using a flatMap() in the way I am using, I am
> better off using :
> * a fold (init, fold_func, window_func) first and then
> * map to a different type of my choice, inside the window_func,
> parameterised above
>
> I hope I am correct. If so, you don't need to spend time to comment;
> ☺otherwise, please give a hint.
>
> -- Nirmalya
>
> -
>
> On Thu, Feb 16, 2017 at 4:12 PM, Aljoscha Krettek [via Apache Flink User
> Mailing List archive.] <[hidden email]
> <http:///user/SendEmail.jtp?type=node&node=11677&i=0>> wrote:
>
> Hi,
> you would indeed use apply(), or better fold(<initial_value>,
> <fold_function>, <window_function>) to map the result of folding your
> window to some other data type. If you will, a WindowFunction allows
> "mapping" the result of your windowing to a different type.
>
> Best,
> Aljoscha
>
> On Wed, 15 Feb 2017 at 06:14 nsengupta <[hidden email]
> <http:///user/SendEmail.jtp?type=node&node=11665&i=0>> wrote:
>
> I have gone through this post
> <
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/WindowedStream-operation-questions-td6006.html
> >
> , where Aljoscha explains that /mapping/ on WindowedStream is /not/
> allowed.
>
> So, I think I haven't asked the question properly. Here is (hopefully) a
> better and easier version:
>
> 1. I begin with records of type RawMITSIMTuple.
> 2. When I group them using a Window, I get an
> AllWindowedStream[RawMITSIMTuple].
> 3. I /fold/ the tuples obtained in the Window, which gives me a
> DataStream[Vector[RawMITSIMTuple].
> 4. What I need is a DataStream[PositionReport]. So, I need to flatMap
> the
> output of previous step, where I first get hold of each of the
> RawMITSIMTuple and map that to PositionReport.
>
> val positionReportStream = this
> .readRawMITSIMTuplesInjected(envDefault,args(0))
> .assignAscendingTimestamps(e => e.timeOfReport)
> .windowAll(TumblingEventTimeWindows.of(Time.seconds(30)))
> .fold(Vector[RawMITSIMTuple]())((collectorBin,rawRecord) => {
> collectorBin :+ rawRecord)
> })
> .flatMap(r => r.map(e => this.preparePositionReport(e)))
>
> This gives me what I want, but I feel this is verbose and inefficient. Am I
> thinking correctly? If so, what is a better idiom to use in such cases?
>
> -- Nirmalya
>
>
>
>
> --
> View this message in context:
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Clarification-use-of-AllWindowedStream-apply-function-tp11627p11630.html
> Sent from the Apache Flink User Mailing List archive. mailing list archive
> at Nabble.com.
>
>
>
> ------------------------------
> If you reply to this email, your message will be added to the discussion
> below:
>
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Clarification-use-of-AllWindowedStream-apply-function-tp11627p11665.html
> To unsubscribe from Clarification: use of AllWindowedStream.apply()
> function, click here.
> NAML
> <http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/template/NamlServlet.jtp?macro=macro_viewer&id=instant_html%21nabble%3Aemail.naml&base=nabble.naml.namespaces.BasicNamespace-nabble.view.web.template.NabbleNamespace-nabble.view.web.template.NodeNamespace&breadcrumbs=notify_subscribers%21nabble%3Aemail.naml-instant_emails%21nabble%3Aemail.naml-send_instant_email%21nabble%3Aemail.naml>
>
>
>
>
> --
> Software Technologist
> http://www.linkedin.com/in/nirmalyasengupta
> "If you have built castles in the air, your work need not be lost. That is
> where they should be.
> Now put the foundation under them."
>
> ------------------------------
> View this message in context: Re: Clarification: use of
> AllWindowedStream.apply() function
> <http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Clarification-use-of-AllWindowedStream-apply-function-tp11627p11677.html>
> Sent from the Apache Flink User Mailing List archive. mailing list archive
> <http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/> at
> Nabble.com.
>
Re: Clarification: use of AllWindowedStream.apply() function
Posted by nsengupta <se...@gmail.com>.
Thanks, Aljoscha for the clarification.
I understand that instead of using a flatMap() in the way I am using, I am
better off using :
* a fold (init, fold_func, window_func) first and then
* map to a different type of my choice, inside the window_func,
parameterised above
I hope I am correct. If so, you don't need to spend time to comment;
☺otherwise, please give a hint.
-- Nirmalya
-
On Thu, Feb 16, 2017 at 4:12 PM, Aljoscha Krettek [via Apache Flink User
Mailing List archive.] <ml...@n4.nabble.com> wrote:
> Hi,
> you would indeed use apply(), or better fold(<initial_value>,
> <fold_function>, <window_function>) to map the result of folding your
> window to some other data type. If you will, a WindowFunction allows
> "mapping" the result of your windowing to a different type.
>
> Best,
> Aljoscha
>
> On Wed, 15 Feb 2017 at 06:14 nsengupta <[hidden email]
> <http:///user/SendEmail.jtp?type=node&node=11665&i=0>> wrote:
>
>> I have gone through this post
>> <http://apache-flink-user-mailing-list-archive.2336050.
>> n4.nabble.com/WindowedStream-operation-questions-td6006.html>
>> , where Aljoscha explains that /mapping/ on WindowedStream is /not/
>> allowed.
>>
>> So, I think I haven't asked the question properly. Here is (hopefully) a
>> better and easier version:
>>
>> 1. I begin with records of type RawMITSIMTuple.
>> 2. When I group them using a Window, I get an
>> AllWindowedStream[RawMITSIMTuple].
>> 3. I /fold/ the tuples obtained in the Window, which gives me a
>> DataStream[Vector[RawMITSIMTuple].
>> 4. What I need is a DataStream[PositionReport]. So, I need to flatMap
>> the
>> output of previous step, where I first get hold of each of the
>> RawMITSIMTuple and map that to PositionReport.
>>
>> val positionReportStream = this
>> .readRawMITSIMTuplesInjected(envDefault,args(0))
>> .assignAscendingTimestamps(e => e.timeOfReport)
>> .windowAll(TumblingEventTimeWindows.of(Time.seconds(30)))
>> .fold(Vector[RawMITSIMTuple]())((collectorBin,rawRecord) => {
>> collectorBin :+ rawRecord)
>> })
>> .flatMap(r => r.map(e => this.preparePositionReport(e)))
>>
>> This gives me what I want, but I feel this is verbose and inefficient. Am
>> I
>> thinking correctly? If so, what is a better idiom to use in such cases?
>>
>> -- Nirmalya
>>
>>
>>
>>
>> --
>> View this message in context: http://apache-flink-user-
>> mailing-list-archive.2336050.n4.nabble.com/Clarification-
>> use-of-AllWindowedStream-apply-function-tp11627p11630.html
>> Sent from the Apache Flink User Mailing List archive. mailing list
>> archive at Nabble.com.
>>
>
>
> ------------------------------
> If you reply to this email, your message will be added to the discussion
> below:
> http://apache-flink-user-mailing-list-archive.2336050.
> n4.nabble.com/Clarification-use-of-AllWindowedStream-
> apply-function-tp11627p11665.html
> To unsubscribe from Clarification: use of AllWindowedStream.apply()
> function, click here
> <http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/template/NamlServlet.jtp?macro=unsubscribe_by_code&node=11627&code=c2VuZ3VwdGEubmlybWFseWFAZ21haWwuY29tfDExNjI3fC01NzQyMjQyNDk=>
> .
> NAML
> <http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/template/NamlServlet.jtp?macro=macro_viewer&id=instant_html%21nabble%3Aemail.naml&base=nabble.naml.namespaces.BasicNamespace-nabble.view.web.template.NabbleNamespace-nabble.view.web.template.NodeNamespace&breadcrumbs=notify_subscribers%21nabble%3Aemail.naml-instant_emails%21nabble%3Aemail.naml-send_instant_email%21nabble%3Aemail.naml>
>
--
Software Technologist
http://www.linkedin.com/in/nirmalyasengupta
"If you have built castles in the air, your work need not be lost. That is
where they should be.
Now put the foundation under them."
--
View this message in context: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Clarification-use-of-AllWindowedStream-apply-function-tp11627p11677.html
Sent from the Apache Flink User Mailing List archive. mailing list archive at Nabble.com.
Re: Clarification: use of AllWindowedStream.apply() function
Posted by Aljoscha Krettek <al...@apache.org>.
Hi,
you would indeed use apply(), or better fold(<initial_value>,
<fold_function>, <window_function>) to map the result of folding your
window to some other data type. If you will, a WindowFunction allows
"mapping" the result of your windowing to a different type.
Best,
Aljoscha
On Wed, 15 Feb 2017 at 06:14 nsengupta <se...@gmail.com> wrote:
> I have gone through this post
> <
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/WindowedStream-operation-questions-td6006.html
> >
> , where Aljoscha explains that /mapping/ on WindowedStream is /not/
> allowed.
>
> So, I think I haven't asked the question properly. Here is (hopefully) a
> better and easier version:
>
> 1. I begin with records of type RawMITSIMTuple.
> 2. When I group them using a Window, I get an
> AllWindowedStream[RawMITSIMTuple].
> 3. I /fold/ the tuples obtained in the Window, which gives me a
> DataStream[Vector[RawMITSIMTuple].
> 4. What I need is a DataStream[PositionReport]. So, I need to flatMap
> the
> output of previous step, where I first get hold of each of the
> RawMITSIMTuple and map that to PositionReport.
>
> val positionReportStream = this
> .readRawMITSIMTuplesInjected(envDefault,args(0))
> .assignAscendingTimestamps(e => e.timeOfReport)
> .windowAll(TumblingEventTimeWindows.of(Time.seconds(30)))
> .fold(Vector[RawMITSIMTuple]())((collectorBin,rawRecord) => {
> collectorBin :+ rawRecord)
> })
> .flatMap(r => r.map(e => this.preparePositionReport(e)))
>
> This gives me what I want, but I feel this is verbose and inefficient. Am I
> thinking correctly? If so, what is a better idiom to use in such cases?
>
> -- Nirmalya
>
>
>
>
> --
> View this message in context:
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Clarification-use-of-AllWindowedStream-apply-function-tp11627p11630.html
> Sent from the Apache Flink User Mailing List archive. mailing list archive
> at Nabble.com.
>
Re: Clarification: use of AllWindowedStream.apply() function
Posted by nsengupta <se...@gmail.com>.
I have gone through this post
<http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/WindowedStream-operation-questions-td6006.html>
, where Aljoscha explains that /mapping/ on WindowedStream is /not/ allowed.
So, I think I haven't asked the question properly. Here is (hopefully) a
better and easier version:
1. I begin with records of type RawMITSIMTuple.
2. When I group them using a Window, I get an
AllWindowedStream[RawMITSIMTuple].
3. I /fold/ the tuples obtained in the Window, which gives me a
DataStream[Vector[RawMITSIMTuple].
4. What I need is a DataStream[PositionReport]. So, I need to flatMap the
output of previous step, where I first get hold of each of the
RawMITSIMTuple and map that to PositionReport.
val positionReportStream = this
.readRawMITSIMTuplesInjected(envDefault,args(0))
.assignAscendingTimestamps(e => e.timeOfReport)
.windowAll(TumblingEventTimeWindows.of(Time.seconds(30)))
.fold(Vector[RawMITSIMTuple]())((collectorBin,rawRecord) => {
collectorBin :+ rawRecord)
})
.flatMap(r => r.map(e => this.preparePositionReport(e)))
This gives me what I want, but I feel this is verbose and inefficient. Am I
thinking correctly? If so, what is a better idiom to use in such cases?
-- Nirmalya
--
View this message in context: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Clarification-use-of-AllWindowedStream-apply-function-tp11627p11630.html
Sent from the Apache Flink User Mailing List archive. mailing list archive at Nabble.com.