You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@streampipes.apache.org by GitBox <gi...@apache.org> on 2021/03/29 12:34:38 UTC
[GitHub] [incubator-streampipes-extensions] Samarth08 opened a new pull request #39: Streampipes 323:Implemented Merge by Schema
Samarth08 opened a new pull request #39:
URL: https://github.com/apache/incubator-streampipes-extensions/pull/39
Fixes: https://issues.apache.org/jira/browse/STREAMPIPES-323
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-streampipes-extensions] Samarth08 commented on pull request #39: [STREAMPIPES-323] Implemented Merge by Schema
Posted by GitBox <gi...@apache.org>.
Samarth08 commented on pull request #39:
URL: https://github.com/apache/incubator-streampipes-extensions/pull/39#issuecomment-820089955
@bossenti I have addressed the review comments.Please have a look.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [GitHub] [incubator-streampipes-extensions] Samarth08
opened a new pull request #39: Streampipes 323:Implemented Merge by
Schema
Posted by udeho <ud...@student.kit.edu>.
sorry, I meant SchemaInfo instead of EventSchema
On März 30 2021, at 4:07 pm, udeho <ud...@student.kit.edu> wrote:
> Hi Samarth,
>
> welcome to our mailing list and thank you for your contribution!
> I just had a look at your PR.
> Just to get it right, it is your idea to check for each incoming event if there are already one or more events in the StreamBuffer that can be merged with the incoming one? And if there are multiple matches, multiple events are emitted?
> I'm not quite sure, whether we should provide the buffering per default or only by choice because leads to big confusion if it's not expected. At least, we should limit buffering to when a new event arrives for the buffered stream, so that un-synchronized streams can still be merged.
>
> @all, how would you handle the stream buffering?
> In the declare model you have to specify the two required inputStreams. For example this could look like:
>
> .requiredStream(StreamRequirementsBuilder
> .create()
> .requiredProperty(EpRequirements.anyProperty())
> .build())
> .requiredStream(StreamRequirementsBuilder
> .create()
> .requiredProperty(EpRequirements.anyProperty())
> .build())
> Also, the EventSchema does not yet override the equals method and therefore the comparison does not work properly. If you want, you can try to implement that as well.
>
> Apart from this general things, your provided code looks very good.
> There are some minor things that we can address once we have agreed on this point, if that's okay with you?
>
> Best
> Tim
>
> On März 29 2021, at 2:57 pm, Samarth Sah <s....@gmail.com> wrote:
> > Hi All,
> >
> > I am new to Streampipes.
> > I have raised the pull request:
> > https://github.com/apache/incubator-streampipes-extensions/pull/39,
> > regarding the issue https://issues.apache.org/jira/browse/STREAMPIPES-323.
> >
> > Please let me know if I am in the right direction.
> > Thank you,
> > Samarth
> >
> >
> >
> > On Mon, Mar 29, 2021 at 6:04 PM GitBox <gi...@apache.org> wrote:
> > >
> > > Samarth08 opened a new pull request #39:
> > > URL: https://github.com/apache/incubator-streampipes-extensions/pull/39
> > >
> > >
> > > Fixes: https://issues.apache.org/jira/browse/STREAMPIPES-323
> > >
> > >
> > >
> > > --
> > > This is an automated message from the Apache Git Service.
> > > To respond to the message, please log on to GitHub and use the
> > > URL above to go to the specific comment.
> > >
> > > For queries about this service, please contact Infrastructure at:
> > > users@infra.apache.org
> > >
> > >
> > >
> >
>
Re: [GitHub] [incubator-streampipes-extensions] Samarth08 opened a
new pull request #39: Streampipes 323:Implemented Merge by Schema
Posted by Samarth Sah <s....@gmail.com>.
Hi Tim,
I have raised a PR for overriding the equals method for schema Info.
https://github.com/apache/incubator-streampipes/pull/34
Please Check.
Thank You
Samarth
On Thu, Apr 1, 2021 at 6:28 PM udeho <ud...@student.kit.edu> wrote:
> Hi Samarth,
>
> I had a quick discussion with Philipp who initially came up with the idea
> of the MergeBySchema processor.
> The current solution is quite similar to already existing process elements.
> So I would ask you to implement it as follows:
> - the processing element requires to input streams
> - inside the onInvocation method both SchemaInfos are compared and if they
> don't match (are the same) a SpRuntimeException is thrown. Here I would say
> that two SchemaInfos are equal if their EventSchemas contain the same
> EventProperties. Ideally, we would already check the SchemaInfos in the UI
> component, but unfortunately we don't have corresponding UI components yet.
> - the onEvent method simply forwards each event without any further
> manipulations
>
> Sorry for the confusion about the problem to be solved, I should have
> written a clearer description of the processor.
> Best
> Tim
> On März 31 2021, at 6:41 am, Samarth Sah <s....@gmail.com> wrote:
> > Hi Tim/Team,
> >
> > Please let me know the parameters to be used in the overriding
> > equals method for SchemaInfo.
> >
> > For emitting of multiple events can we try the below logic:
> > for (Event e0 : this.streamBufferS0.getList()) {
> > boolean atLeastOneMerge=false;
> > for (Event e1 : this.streamBufferS1.getList()) {
> > if (e0.getSchemaInfo().equals(e1.getSchemaInfo())) {
> > atLeastOneMerge=true;
> > Event resultingEvent = mergeEvents(e0, e1);
> > spOutputCollector.collect(resultingEvent);
> > this.streamBufferS1.removeOldEvent(e1);
> > }
> > }
> > if(atLeastOneMerge)
> > this.streamBufferS1.removeOldEvent(e0);
> > }
> >
> >
> > Thank You
> > Samarth
> >
> >
> > On Tue, Mar 30, 2021 at 8:18 PM Samarth Sah <s....@gmail.com> wrote:
> > > Hi Tim,
> > >
> > > Yes, the idea is to check for each incoming event if there are already
> one
> > > or more events in the StreamBuffer that can be merged with the
> incoming one.
> > >
> > > However I am not sure about emitting the events in case of multiple
> > > matches.Do you want me to stop checking after one match?Since that
> event is
> > > already merged.
> > >
> > > I will try to override the equals methods for SchemaInfo.
> > >
> > > Thank You,
> > > Samarth
> > >
> > >
> > >
> > >
> > >
> > > On Tue, Mar 30, 2021 at 7:38 PM udeho <ud...@student.kit.edu> wrote:
> > >
> > >> Hi Samarth,
> > >>
> > >> welcome to our mailing list and thank you for your contribution!
> > >> I just had a look at your PR.
> > >> Just to get it right, it is your idea to check for each incoming
> event if
> > >> there are already one or more events in the StreamBuffer that can be
> merged
> > >> with the incoming one? And if there are multiple matches, multiple
> events
> > >> are emitted?
> > >> I'm not quite sure, whether we should provide the buffering per
> default
> > >> or only by choice because leads to big confusion if it's not
> expected. At
> > >> least, we should limit buffering to when a new event arrives for the
> > >> buffered stream, so that un-synchronized streams can still be merged.
> > >>
> > >> @all, how would you handle the stream buffering?
> > >> In the declare model you have to specify the two required
> inputStreams.
> > >> For example this could look like:
> > >>
> > >> .requiredStream(StreamRequirementsBuilder
> > >> .create()
> > >> .requiredProperty(EpRequirements.anyProperty())
> > >> .build())
> > >> .requiredStream(StreamRequirementsBuilder
> > >> .create()
> > >> .requiredProperty(EpRequirements.anyProperty())
> > >> .build())
> > >> Also, the EventSchema does not yet override the equals method and
> > >> therefore the comparison does not work properly. If you want, you can
> try
> > >> to implement that as well.
> > >>
> > >> Apart from this general things, your provided code looks very good.
> > >> There are some minor things that we can address once we have agreed on
> > >> this point, if that's okay with you?
> > >>
> > >> Best
> > >> Tim
> > >>
> > >> On März 29 2021, at 2:57 pm, Samarth Sah <s....@gmail.com> wrote:
> > >> > Hi All,
> > >> >
> > >> > I am new to Streampipes.
> > >> > I have raised the pull request:
> > >> > https://github.com/apache/incubator-streampipes-extensions/pull/39,
> > >> > regarding the issue
> > >> https://issues.apache.org/jira/browse/STREAMPIPES-323.
> > >> >
> > >> > Please let me know if I am in the right direction.
> > >> > Thank you,
> > >> > Samarth
> > >> >
> > >> >
> > >> >
> > >> > On Mon, Mar 29, 2021 at 6:04 PM GitBox <gi...@apache.org> wrote:
> > >> > >
> > >> > > Samarth08 opened a new pull request #39:
> > >> > > URL:
> > >> https://github.com/apache/incubator-streampipes-extensions/pull/39
> > >> > >
> > >> > >
> > >> > > Fixes: https://issues.apache.org/jira/browse/STREAMPIPES-323
> > >> > >
> > >> > >
> > >> > >
> > >> > > --
> > >> > > This is an automated message from the Apache Git Service.
> > >> > > To respond to the message, please log on to GitHub and use the
> > >> > > URL above to go to the specific comment.
> > >> > >
> > >> > > For queries about this service, please contact Infrastructure at:
> > >> > > users@infra.apache.org
> > >> > >
> > >> > >
> > >> > >
> > >> >
> > >>
> > >>
> >
>
>
Re: [GitHub] [incubator-streampipes-extensions] Samarth08 opened a
new pull request #39: Streampipes 323:Implemented Merge by Schema
Posted by Samarth Sah <s....@gmail.com>.
Hi Tim/Team,
I have made the above mentioned changes in the extensions project.
https://github.com/apache/incubator-streampipes-extensions/pull/39
Now there are two PR's open for this issue: one has the overriding equals
logic(https://github.com/apache/incubator-streampipes/pull/34) the other
one has the actual implementation.
Thank You
Samarth
On Thu, Apr 1, 2021 at 6:28 PM udeho <ud...@student.kit.edu> wrote:
> Hi Samarth,
>
> I had a quick discussion with Philipp who initially came up with the idea
> of the MergeBySchema processor.
> The current solution is quite similar to already existing process elements.
> So I would ask you to implement it as follows:
> - the processing element requires to input streams
> - inside the onInvocation method both SchemaInfos are compared and if they
> don't match (are the same) a SpRuntimeException is thrown. Here I would say
> that two SchemaInfos are equal if their EventSchemas contain the same
> EventProperties. Ideally, we would already check the SchemaInfos in the UI
> component, but unfortunately we don't have corresponding UI components yet.
> - the onEvent method simply forwards each event without any further
> manipulations
>
> Sorry for the confusion about the problem to be solved, I should have
> written a clearer description of the processor.
> Best
> Tim
> On März 31 2021, at 6:41 am, Samarth Sah <s....@gmail.com> wrote:
> > Hi Tim/Team,
> >
> > Please let me know the parameters to be used in the overriding
> > equals method for SchemaInfo.
> >
> > For emitting of multiple events can we try the below logic:
> > for (Event e0 : this.streamBufferS0.getList()) {
> > boolean atLeastOneMerge=false;
> > for (Event e1 : this.streamBufferS1.getList()) {
> > if (e0.getSchemaInfo().equals(e1.getSchemaInfo())) {
> > atLeastOneMerge=true;
> > Event resultingEvent = mergeEvents(e0, e1);
> > spOutputCollector.collect(resultingEvent);
> > this.streamBufferS1.removeOldEvent(e1);
> > }
> > }
> > if(atLeastOneMerge)
> > this.streamBufferS1.removeOldEvent(e0);
> > }
> >
> >
> > Thank You
> > Samarth
> >
> >
> > On Tue, Mar 30, 2021 at 8:18 PM Samarth Sah <s....@gmail.com> wrote:
> > > Hi Tim,
> > >
> > > Yes, the idea is to check for each incoming event if there are already
> one
> > > or more events in the StreamBuffer that can be merged with the
> incoming one.
> > >
> > > However I am not sure about emitting the events in case of multiple
> > > matches.Do you want me to stop checking after one match?Since that
> event is
> > > already merged.
> > >
> > > I will try to override the equals methods for SchemaInfo.
> > >
> > > Thank You,
> > > Samarth
> > >
> > >
> > >
> > >
> > >
> > > On Tue, Mar 30, 2021 at 7:38 PM udeho <ud...@student.kit.edu> wrote:
> > >
> > >> Hi Samarth,
> > >>
> > >> welcome to our mailing list and thank you for your contribution!
> > >> I just had a look at your PR.
> > >> Just to get it right, it is your idea to check for each incoming
> event if
> > >> there are already one or more events in the StreamBuffer that can be
> merged
> > >> with the incoming one? And if there are multiple matches, multiple
> events
> > >> are emitted?
> > >> I'm not quite sure, whether we should provide the buffering per
> default
> > >> or only by choice because leads to big confusion if it's not
> expected. At
> > >> least, we should limit buffering to when a new event arrives for the
> > >> buffered stream, so that un-synchronized streams can still be merged.
> > >>
> > >> @all, how would you handle the stream buffering?
> > >> In the declare model you have to specify the two required
> inputStreams.
> > >> For example this could look like:
> > >>
> > >> .requiredStream(StreamRequirementsBuilder
> > >> .create()
> > >> .requiredProperty(EpRequirements.anyProperty())
> > >> .build())
> > >> .requiredStream(StreamRequirementsBuilder
> > >> .create()
> > >> .requiredProperty(EpRequirements.anyProperty())
> > >> .build())
> > >> Also, the EventSchema does not yet override the equals method and
> > >> therefore the comparison does not work properly. If you want, you can
> try
> > >> to implement that as well.
> > >>
> > >> Apart from this general things, your provided code looks very good.
> > >> There are some minor things that we can address once we have agreed on
> > >> this point, if that's okay with you?
> > >>
> > >> Best
> > >> Tim
> > >>
> > >> On März 29 2021, at 2:57 pm, Samarth Sah <s....@gmail.com> wrote:
> > >> > Hi All,
> > >> >
> > >> > I am new to Streampipes.
> > >> > I have raised the pull request:
> > >> > https://github.com/apache/incubator-streampipes-extensions/pull/39,
> > >> > regarding the issue
> > >> https://issues.apache.org/jira/browse/STREAMPIPES-323.
> > >> >
> > >> > Please let me know if I am in the right direction.
> > >> > Thank you,
> > >> > Samarth
> > >> >
> > >> >
> > >> >
> > >> > On Mon, Mar 29, 2021 at 6:04 PM GitBox <gi...@apache.org> wrote:
> > >> > >
> > >> > > Samarth08 opened a new pull request #39:
> > >> > > URL:
> > >> https://github.com/apache/incubator-streampipes-extensions/pull/39
> > >> > >
> > >> > >
> > >> > > Fixes: https://issues.apache.org/jira/browse/STREAMPIPES-323
> > >> > >
> > >> > >
> > >> > >
> > >> > > --
> > >> > > This is an automated message from the Apache Git Service.
> > >> > > To respond to the message, please log on to GitHub and use the
> > >> > > URL above to go to the specific comment.
> > >> > >
> > >> > > For queries about this service, please contact Infrastructure at:
> > >> > > users@infra.apache.org
> > >> > >
> > >> > >
> > >> > >
> > >> >
> > >>
> > >>
> >
>
>
Re: [GitHub] [incubator-streampipes-extensions] Samarth08
opened a new pull request #39: Streampipes 323:Implemented Merge by
Schema
Posted by udeho <ud...@student.kit.edu>.
Hi Samarth,
I had a quick discussion with Philipp who initially came up with the idea of the MergeBySchema processor.
The current solution is quite similar to already existing process elements.
So I would ask you to implement it as follows:
- the processing element requires to input streams
- inside the onInvocation method both SchemaInfos are compared and if they don't match (are the same) a SpRuntimeException is thrown. Here I would say that two SchemaInfos are equal if their EventSchemas contain the same EventProperties. Ideally, we would already check the SchemaInfos in the UI component, but unfortunately we don't have corresponding UI components yet.
- the onEvent method simply forwards each event without any further manipulations
Sorry for the confusion about the problem to be solved, I should have written a clearer description of the processor.
Best
Tim
On März 31 2021, at 6:41 am, Samarth Sah <s....@gmail.com> wrote:
> Hi Tim/Team,
>
> Please let me know the parameters to be used in the overriding
> equals method for SchemaInfo.
>
> For emitting of multiple events can we try the below logic:
> for (Event e0 : this.streamBufferS0.getList()) {
> boolean atLeastOneMerge=false;
> for (Event e1 : this.streamBufferS1.getList()) {
> if (e0.getSchemaInfo().equals(e1.getSchemaInfo())) {
> atLeastOneMerge=true;
> Event resultingEvent = mergeEvents(e0, e1);
> spOutputCollector.collect(resultingEvent);
> this.streamBufferS1.removeOldEvent(e1);
> }
> }
> if(atLeastOneMerge)
> this.streamBufferS1.removeOldEvent(e0);
> }
>
>
> Thank You
> Samarth
>
>
> On Tue, Mar 30, 2021 at 8:18 PM Samarth Sah <s....@gmail.com> wrote:
> > Hi Tim,
> >
> > Yes, the idea is to check for each incoming event if there are already one
> > or more events in the StreamBuffer that can be merged with the incoming one.
> >
> > However I am not sure about emitting the events in case of multiple
> > matches.Do you want me to stop checking after one match?Since that event is
> > already merged.
> >
> > I will try to override the equals methods for SchemaInfo.
> >
> > Thank You,
> > Samarth
> >
> >
> >
> >
> >
> > On Tue, Mar 30, 2021 at 7:38 PM udeho <ud...@student.kit.edu> wrote:
> >
> >> Hi Samarth,
> >>
> >> welcome to our mailing list and thank you for your contribution!
> >> I just had a look at your PR.
> >> Just to get it right, it is your idea to check for each incoming event if
> >> there are already one or more events in the StreamBuffer that can be merged
> >> with the incoming one? And if there are multiple matches, multiple events
> >> are emitted?
> >> I'm not quite sure, whether we should provide the buffering per default
> >> or only by choice because leads to big confusion if it's not expected. At
> >> least, we should limit buffering to when a new event arrives for the
> >> buffered stream, so that un-synchronized streams can still be merged.
> >>
> >> @all, how would you handle the stream buffering?
> >> In the declare model you have to specify the two required inputStreams.
> >> For example this could look like:
> >>
> >> .requiredStream(StreamRequirementsBuilder
> >> .create()
> >> .requiredProperty(EpRequirements.anyProperty())
> >> .build())
> >> .requiredStream(StreamRequirementsBuilder
> >> .create()
> >> .requiredProperty(EpRequirements.anyProperty())
> >> .build())
> >> Also, the EventSchema does not yet override the equals method and
> >> therefore the comparison does not work properly. If you want, you can try
> >> to implement that as well.
> >>
> >> Apart from this general things, your provided code looks very good.
> >> There are some minor things that we can address once we have agreed on
> >> this point, if that's okay with you?
> >>
> >> Best
> >> Tim
> >>
> >> On März 29 2021, at 2:57 pm, Samarth Sah <s....@gmail.com> wrote:
> >> > Hi All,
> >> >
> >> > I am new to Streampipes.
> >> > I have raised the pull request:
> >> > https://github.com/apache/incubator-streampipes-extensions/pull/39,
> >> > regarding the issue
> >> https://issues.apache.org/jira/browse/STREAMPIPES-323.
> >> >
> >> > Please let me know if I am in the right direction.
> >> > Thank you,
> >> > Samarth
> >> >
> >> >
> >> >
> >> > On Mon, Mar 29, 2021 at 6:04 PM GitBox <gi...@apache.org> wrote:
> >> > >
> >> > > Samarth08 opened a new pull request #39:
> >> > > URL:
> >> https://github.com/apache/incubator-streampipes-extensions/pull/39
> >> > >
> >> > >
> >> > > Fixes: https://issues.apache.org/jira/browse/STREAMPIPES-323
> >> > >
> >> > >
> >> > >
> >> > > --
> >> > > This is an automated message from the Apache Git Service.
> >> > > To respond to the message, please log on to GitHub and use the
> >> > > URL above to go to the specific comment.
> >> > >
> >> > > For queries about this service, please contact Infrastructure at:
> >> > > users@infra.apache.org
> >> > >
> >> > >
> >> > >
> >> >
> >>
> >>
>
Re: [GitHub] [incubator-streampipes-extensions] Samarth08 opened a
new pull request #39: Streampipes 323:Implemented Merge by Schema
Posted by Samarth Sah <s....@gmail.com>.
Hi Tim/Team,
Please let me know the parameters to be used in the overriding
equals method for SchemaInfo.
For emitting of multiple events can we try the below logic:
for (Event e0 : this.streamBufferS0.getList()) {
boolean atLeastOneMerge=false;
for (Event e1 : this.streamBufferS1.getList()) {
if (e0.getSchemaInfo().equals(e1.getSchemaInfo())) {
atLeastOneMerge=true;
Event resultingEvent = mergeEvents(e0, e1);
spOutputCollector.collect(resultingEvent);
this.streamBufferS1.removeOldEvent(e1);
}
}
if(atLeastOneMerge)
this.streamBufferS1.removeOldEvent(e0);
}
Thank You
Samarth
On Tue, Mar 30, 2021 at 8:18 PM Samarth Sah <s....@gmail.com> wrote:
> Hi Tim,
>
> Yes, the idea is to check for each incoming event if there are already one
> or more events in the StreamBuffer that can be merged with the incoming one.
>
> However I am not sure about emitting the events in case of multiple
> matches.Do you want me to stop checking after one match?Since that event is
> already merged.
>
> I will try to override the equals methods for SchemaInfo.
>
> Thank You,
> Samarth
>
>
>
>
>
> On Tue, Mar 30, 2021 at 7:38 PM udeho <ud...@student.kit.edu> wrote:
>
>> Hi Samarth,
>>
>> welcome to our mailing list and thank you for your contribution!
>> I just had a look at your PR.
>> Just to get it right, it is your idea to check for each incoming event if
>> there are already one or more events in the StreamBuffer that can be merged
>> with the incoming one? And if there are multiple matches, multiple events
>> are emitted?
>> I'm not quite sure, whether we should provide the buffering per default
>> or only by choice because leads to big confusion if it's not expected. At
>> least, we should limit buffering to when a new event arrives for the
>> buffered stream, so that un-synchronized streams can still be merged.
>>
>> @all, how would you handle the stream buffering?
>> In the declare model you have to specify the two required inputStreams.
>> For example this could look like:
>>
>> .requiredStream(StreamRequirementsBuilder
>> .create()
>> .requiredProperty(EpRequirements.anyProperty())
>> .build())
>> .requiredStream(StreamRequirementsBuilder
>> .create()
>> .requiredProperty(EpRequirements.anyProperty())
>> .build())
>> Also, the EventSchema does not yet override the equals method and
>> therefore the comparison does not work properly. If you want, you can try
>> to implement that as well.
>>
>> Apart from this general things, your provided code looks very good.
>> There are some minor things that we can address once we have agreed on
>> this point, if that's okay with you?
>>
>> Best
>> Tim
>>
>> On März 29 2021, at 2:57 pm, Samarth Sah <s....@gmail.com> wrote:
>> > Hi All,
>> >
>> > I am new to Streampipes.
>> > I have raised the pull request:
>> > https://github.com/apache/incubator-streampipes-extensions/pull/39,
>> > regarding the issue
>> https://issues.apache.org/jira/browse/STREAMPIPES-323.
>> >
>> > Please let me know if I am in the right direction.
>> > Thank you,
>> > Samarth
>> >
>> >
>> >
>> > On Mon, Mar 29, 2021 at 6:04 PM GitBox <gi...@apache.org> wrote:
>> > >
>> > > Samarth08 opened a new pull request #39:
>> > > URL:
>> https://github.com/apache/incubator-streampipes-extensions/pull/39
>> > >
>> > >
>> > > Fixes: https://issues.apache.org/jira/browse/STREAMPIPES-323
>> > >
>> > >
>> > >
>> > > --
>> > > This is an automated message from the Apache Git Service.
>> > > To respond to the message, please log on to GitHub and use the
>> > > URL above to go to the specific comment.
>> > >
>> > > For queries about this service, please contact Infrastructure at:
>> > > users@infra.apache.org
>> > >
>> > >
>> > >
>> >
>>
>>
Re: [GitHub] [incubator-streampipes-extensions] Samarth08 opened a
new pull request #39: Streampipes 323:Implemented Merge by Schema
Posted by Samarth Sah <s....@gmail.com>.
Hi Tim,
Yes, the idea is to check for each incoming event if there are already one
or more events in the StreamBuffer that can be merged with the incoming one.
However I am not sure about emitting the events in case of multiple
matches.Do you want me to stop checking after one match?Since that event is
already merged.
I will try to override the equals methods for SchemaInfo.
Thank You,
Samarth
On Tue, Mar 30, 2021 at 7:38 PM udeho <ud...@student.kit.edu> wrote:
> Hi Samarth,
>
> welcome to our mailing list and thank you for your contribution!
> I just had a look at your PR.
> Just to get it right, it is your idea to check for each incoming event if
> there are already one or more events in the StreamBuffer that can be merged
> with the incoming one? And if there are multiple matches, multiple events
> are emitted?
> I'm not quite sure, whether we should provide the buffering per default or
> only by choice because leads to big confusion if it's not expected. At
> least, we should limit buffering to when a new event arrives for the
> buffered stream, so that un-synchronized streams can still be merged.
>
> @all, how would you handle the stream buffering?
> In the declare model you have to specify the two required inputStreams.
> For example this could look like:
>
> .requiredStream(StreamRequirementsBuilder
> .create()
> .requiredProperty(EpRequirements.anyProperty())
> .build())
> .requiredStream(StreamRequirementsBuilder
> .create()
> .requiredProperty(EpRequirements.anyProperty())
> .build())
> Also, the EventSchema does not yet override the equals method and
> therefore the comparison does not work properly. If you want, you can try
> to implement that as well.
>
> Apart from this general things, your provided code looks very good.
> There are some minor things that we can address once we have agreed on
> this point, if that's okay with you?
>
> Best
> Tim
>
> On März 29 2021, at 2:57 pm, Samarth Sah <s....@gmail.com> wrote:
> > Hi All,
> >
> > I am new to Streampipes.
> > I have raised the pull request:
> > https://github.com/apache/incubator-streampipes-extensions/pull/39,
> > regarding the issue
> https://issues.apache.org/jira/browse/STREAMPIPES-323.
> >
> > Please let me know if I am in the right direction.
> > Thank you,
> > Samarth
> >
> >
> >
> > On Mon, Mar 29, 2021 at 6:04 PM GitBox <gi...@apache.org> wrote:
> > >
> > > Samarth08 opened a new pull request #39:
> > > URL:
> https://github.com/apache/incubator-streampipes-extensions/pull/39
> > >
> > >
> > > Fixes: https://issues.apache.org/jira/browse/STREAMPIPES-323
> > >
> > >
> > >
> > > --
> > > This is an automated message from the Apache Git Service.
> > > To respond to the message, please log on to GitHub and use the
> > > URL above to go to the specific comment.
> > >
> > > For queries about this service, please contact Infrastructure at:
> > > users@infra.apache.org
> > >
> > >
> > >
> >
>
>
Re: [GitHub] [incubator-streampipes-extensions] Samarth08
opened a new pull request #39: Streampipes 323:Implemented Merge by
Schema
Posted by udeho <ud...@student.kit.edu>.
Hi Samarth,
welcome to our mailing list and thank you for your contribution!
I just had a look at your PR.
Just to get it right, it is your idea to check for each incoming event if there are already one or more events in the StreamBuffer that can be merged with the incoming one? And if there are multiple matches, multiple events are emitted?
I'm not quite sure, whether we should provide the buffering per default or only by choice because leads to big confusion if it's not expected. At least, we should limit buffering to when a new event arrives for the buffered stream, so that un-synchronized streams can still be merged.
@all, how would you handle the stream buffering?
In the declare model you have to specify the two required inputStreams. For example this could look like:
.requiredStream(StreamRequirementsBuilder
.create()
.requiredProperty(EpRequirements.anyProperty())
.build())
.requiredStream(StreamRequirementsBuilder
.create()
.requiredProperty(EpRequirements.anyProperty())
.build())
Also, the EventSchema does not yet override the equals method and therefore the comparison does not work properly. If you want, you can try to implement that as well.
Apart from this general things, your provided code looks very good.
There are some minor things that we can address once we have agreed on this point, if that's okay with you?
Best
Tim
On März 29 2021, at 2:57 pm, Samarth Sah <s....@gmail.com> wrote:
> Hi All,
>
> I am new to Streampipes.
> I have raised the pull request:
> https://github.com/apache/incubator-streampipes-extensions/pull/39,
> regarding the issue https://issues.apache.org/jira/browse/STREAMPIPES-323.
>
> Please let me know if I am in the right direction.
> Thank you,
> Samarth
>
>
>
> On Mon, Mar 29, 2021 at 6:04 PM GitBox <gi...@apache.org> wrote:
> >
> > Samarth08 opened a new pull request #39:
> > URL: https://github.com/apache/incubator-streampipes-extensions/pull/39
> >
> >
> > Fixes: https://issues.apache.org/jira/browse/STREAMPIPES-323
> >
> >
> >
> > --
> > This is an automated message from the Apache Git Service.
> > To respond to the message, please log on to GitHub and use the
> > URL above to go to the specific comment.
> >
> > For queries about this service, please contact Infrastructure at:
> > users@infra.apache.org
> >
> >
> >
>
Re: [GitHub] [incubator-streampipes-extensions] Samarth08 opened a
new pull request #39: Streampipes 323:Implemented Merge by Schema
Posted by Samarth Sah <s....@gmail.com>.
Hi All,
I am new to Streampipes.
I have raised the pull request:
https://github.com/apache/incubator-streampipes-extensions/pull/39,
regarding the issue https://issues.apache.org/jira/browse/STREAMPIPES-323.
Please let me know if I am in the right direction.
Thank you,
Samarth
On Mon, Mar 29, 2021 at 6:04 PM GitBox <gi...@apache.org> wrote:
>
> Samarth08 opened a new pull request #39:
> URL: https://github.com/apache/incubator-streampipes-extensions/pull/39
>
>
> Fixes: https://issues.apache.org/jira/browse/STREAMPIPES-323
>
>
>
> --
> This is an automated message from the Apache Git Service.
> To respond to the message, please log on to GitHub and use the
> URL above to go to the specific comment.
>
> For queries about this service, please contact Infrastructure at:
> users@infra.apache.org
>
>
>
[GitHub] [incubator-streampipes-extensions] bossenti commented on a change in pull request #39: [STREAMPIPES-323] Implemented Merge by Schema
Posted by GitBox <gi...@apache.org>.
bossenti commented on a change in pull request #39:
URL: https://github.com/apache/incubator-streampipes-extensions/pull/39#discussion_r612427390
##########
File path: streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/schema/MergeBySchemaProcessor.java
##########
@@ -0,0 +1,60 @@
+package org.apache.streampipes.processors.filters.jvm.processor.schema;
+
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.model.DataProcessorType;
+import org.apache.streampipes.model.graph.DataProcessorDescription;
+import org.apache.streampipes.model.runtime.Event;
+import org.apache.streampipes.model.schema.EventSchema;
+import org.apache.streampipes.sdk.builder.ProcessingElementBuilder;
+import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
+import org.apache.streampipes.sdk.helpers.EpRequirements;
+import org.apache.streampipes.sdk.helpers.Locales;
+import org.apache.streampipes.sdk.helpers.OutputStrategies;
+import org.apache.streampipes.sdk.utils.Assets;
+import org.apache.streampipes.wrapper.context.EventProcessorRuntimeContext;
+import org.apache.streampipes.wrapper.routing.SpOutputCollector;
+import org.apache.streampipes.wrapper.standalone.ProcessorParams;
+import org.apache.streampipes.wrapper.standalone.StreamPipesDataProcessor;
+
+
+public class MergeBySchemaProcessor extends StreamPipesDataProcessor {
+
+
+ @Override
+ public DataProcessorDescription declareModel() {
+ return ProcessingElementBuilder.create("org.apache.streampipes.processors.filters.jvm.schema")
+ .category(DataProcessorType.TRANSFORM)
+ .withAssets(Assets.DOCUMENTATION, Assets.ICON)
+ .withLocales(Locales.EN)
Review comment:
use this only if you provide the mentioned assets in the ressource directory
##########
File path: streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/schema/MergeBySchemaProcessor.java
##########
@@ -0,0 +1,60 @@
+package org.apache.streampipes.processors.filters.jvm.processor.schema;
Review comment:
add this processing element to `FiltersJvmInit` and `AllPipelineElementsInit`
##########
File path: streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/schema/MergeBySchemaProcessor.java
##########
@@ -0,0 +1,60 @@
+package org.apache.streampipes.processors.filters.jvm.processor.schema;
+
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.model.DataProcessorType;
+import org.apache.streampipes.model.graph.DataProcessorDescription;
+import org.apache.streampipes.model.runtime.Event;
+import org.apache.streampipes.model.schema.EventSchema;
+import org.apache.streampipes.sdk.builder.ProcessingElementBuilder;
+import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
+import org.apache.streampipes.sdk.helpers.EpRequirements;
+import org.apache.streampipes.sdk.helpers.Locales;
+import org.apache.streampipes.sdk.helpers.OutputStrategies;
+import org.apache.streampipes.sdk.utils.Assets;
+import org.apache.streampipes.wrapper.context.EventProcessorRuntimeContext;
+import org.apache.streampipes.wrapper.routing.SpOutputCollector;
+import org.apache.streampipes.wrapper.standalone.ProcessorParams;
+import org.apache.streampipes.wrapper.standalone.StreamPipesDataProcessor;
+
+
+public class MergeBySchemaProcessor extends StreamPipesDataProcessor {
+
+
+ @Override
+ public DataProcessorDescription declareModel() {
+ return ProcessingElementBuilder.create("org.apache.streampipes.processors.filters.jvm.schema")
+ .category(DataProcessorType.TRANSFORM)
+ .withAssets(Assets.DOCUMENTATION, Assets.ICON)
+ .withLocales(Locales.EN)
+ .outputStrategy(OutputStrategies.custom(true))
Review comment:
```suggestion
.outputStrategy(OutputStrategies.keep())
```
We do not want to change the events and simply keep them as they are.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-streampipes-extensions] bossenti commented on pull request #39: [STREAMPIPES-323] Implemented Merge by Schema
Posted by GitBox <gi...@apache.org>.
bossenti commented on pull request #39:
URL: https://github.com/apache/incubator-streampipes-extensions/pull/39#issuecomment-818719219
Thanks for the revision.
Like in your other PR the commit messages need to be adapted.
Furthermore, the folder in the ressource directory and the `string.en` file is missing.
Please ensure this is named equally to the id speciefied in the `ProcessingElementBuilder`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-streampipes-extensions] bossenti commented on pull request #39: [STREAMPIPES-323] Implemented Merge by Schema
Posted by GitBox <gi...@apache.org>.
bossenti commented on pull request #39:
URL: https://github.com/apache/incubator-streampipes-extensions/pull/39#issuecomment-820552071
PR is merged. thank you very much for your contribution, @Samarth08!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-streampipes-extensions] bossenti merged pull request #39: [STREAMPIPES-323] Implemented Merge by Schema
Posted by GitBox <gi...@apache.org>.
bossenti merged pull request #39:
URL: https://github.com/apache/incubator-streampipes-extensions/pull/39
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-streampipes-extensions] bossenti commented on a change in pull request #39: [STREAMPIPES-323] Implemented Merge by Schema
Posted by GitBox <gi...@apache.org>.
bossenti commented on a change in pull request #39:
URL: https://github.com/apache/incubator-streampipes-extensions/pull/39#discussion_r612453718
##########
File path: streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/schema/MergeBySchemaProcessor.java
##########
@@ -0,0 +1,60 @@
+package org.apache.streampipes.processors.filters.jvm.processor.schema;
Review comment:
apache header is missing
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org