Re: How to cogroup multiple streams?

You could first transform each stream to a common format (in the worst case, an ugly Either-like capturing all possible types), union those streams, and then do a keyBy + window function. This is how coGroup is implemented internally. On 14/02/2022 16:08, Will Lauer wrote: OK, here's w

How to cogroup multiple streams?

OK, here's what I hope is a stupid question: what's the most efficient way to co-group more than 2 DataStreams together? I'm looking at porting a pipeline from pig to flink, and in a couple of places I use Pig's COGROUP functionality to simultaneously group 3 or 4 and sometime

Re: Speeding up CoGroup in batch job

+%, with the average around 50%. But iowait never gets very high. Wondering if CPU is low when a lot of segments are being flushed to disk, and high when a lot of segments are being sorted before being flushed. The main bottleneck is the CoGroup operation, which is in the phase where it's writ

Re: Speeding up CoGroup in batch job

wrote: > Hi all, > > I added a CoGroup to my batch job, and it’s now running much slower, > primarily due to back pressure from the CoGroup operator. > > I assume it’s because this operator is having to sort/buffer-to-disk all > incoming data. Looks like about 1TB from one side o

Speeding up CoGroup in batch job

Hi all, I added a CoGroup to my batch job, and it’s now running much slower, primarily due to back pressure from the CoGroup operator. I assume it’s because this operator is having to sort/buffer-to-disk all incoming data. Looks like about 1TB from one side of the join, currently very little

Re: Getting Window information from coGroup functin

/2020 13:25, Sudan S wrote: > > Hi, > > I have a usecase where i want to join two streams. I am using coGroup for > this > > KeyBuilder leftKey = new > KeyBuilder(jobConfiguration.getConnectStream().getLeftKey()); > KeyBuilder rightKey = new > KeyBuild

Re: Getting Window information from coGroup functin

object you can fetch the information of window, start time, end time etc. From: Jaswin Shah Sent: 04 June 2020 13:45 To: Dawid Wysakowicz ; user@flink.apache.org Cc: Aljoscha Krettek Subject: Re: Getting Window information from coGroup functin I think here apply

Re: Getting Window information from coGroup functin

information from coGroup functin I am afraid there is no way to do that. At least I could not think of a way to do it. Maybe @aljoscha cc'ed could help here. On 29/05/2020 13:25, Sudan S wrote: Hi, I have a usecase where i want to join two streams. I am using coGroup for this KeyBuilder leftKey

Re: Getting Window information from coGroup functin

I am afraid there is no way to do that. At least I could not think of a way to do it. Maybe @aljoscha cc'ed could help here. On 29/05/2020 13:25, Sudan S wrote: > Hi, > > I have a usecase where i want to join two streams. I am using coGroup > for this > > KeyBuilder lef

Getting Window information from coGroup functin

Hi, I have a usecase where i want to join two streams. I am using coGroup for this KeyBuilder leftKey = new KeyBuilder(jobConfiguration.getConnectStream().getLeftKey()); KeyBuilder rightKey = new KeyBuilder(jobConfiguration.getConnectStream().getRightKey()); leftSource.coGroup(rightSource).where

Re: CoGroup SortMerger performance degradation from 1.6.4 - 1.9.1?

Hi, Yes, it is only related to **batch** jobs, but not necessarily only to DataSet API jobs. If you are using for example Blink SQL/Table API to process some bounded data streams (tables), it could also be visible/affected there. If not, I would suggest to start a new user mailing list question

Re: CoGroup SortMerger performance degradation from 1.6.4 - 1.9.1?

Hi, We encountered similar issues that the task manager kept being killed by yarn. - flink 1.9.1 - heap usage is low. But our job is a **streaming** job, so I want to ask if this issue is only related to **batch** job or not? Thanks! Best, Victor yingjie 于2019年11月28日周四 上午11:43写道： > Piotr is

Re: CoGroup SortMerger performance degradation from 1.6.4 - 1.9.1?

Piotr is right, that depend on the data size you are reading and the memory pressure. Those memory occupied by mmapped region can be recycled and used by other processes if memory pressure is high, that is, other process or service on the same node won't be affected because the OS will recycle the

Re: CoGroup SortMerger performance degradation from 1.6.4 - 1.9.1?

> using mmap for data reading? > > > Best, > Jiayi Liao > > Original Message > Sender: yingjie > Recipient: user > Date: Tuesday, Nov 26, 2019 18:10 > Subject: Re: CoGroup SortMerger performance degradatio

Re: CoGroup SortMerger performance degradation from 1.6.4 - 1.9.1?

right about this. @yingjie Do you have any idea how much memory will be stolen from OS when using mmap for data reading? Best, Jiayi Liao Original Message Sender: yingjie Recipient: user Date: Tuesday, Nov 26, 2019 18:10 Subject: Re: CoGroup SortMerger performance degradation from 1.6.4

Re: CoGroup SortMerger performance degradation from 1.6.4 - 1.9.1?

Thanks for the confirmation, I’ve created Jira ticket to track this issue [1] https://issues.apache.org/jira/browse/FLINK-14952 Piotrek > On 26 Nov 2019, at 11:10, yingjie wrote: > > The new BlockingSubpartition implementation in 1.9 uses mm

Re: CoGroup SortMerger performance degradation from 1.6.4 - 1.9.1?

The new BlockingSubpartition implementation in 1.9 uses mmap for data reading by default which means it steals memory from OS. The mmaped region memory is managed by JVM, so there should be no OutOfMemory problem reported by JVM and the OS memory is also not exhausted, so there should be no kernal

Re: CoGroup SortMerger performance degradation from 1.6.4 - 1.9.1?

our testing. Thanks. > > I’m curious – if I’m understanding this change in 1.9 correctly, blocking > result partitions were being written to mmap which in turn resulted in > exhausting container memory? This is why we were seeing failures in our > pipelines which had operato

RE: CoGroup SortMerger performance degradation from 1.6.4 - 1.9.1?

? This is why we were seeing failures in our pipelines which had operators which fed into a CoGroup? // ah From: Zhijiang Sent: Thursday, November 21, 2019 9:48 PM To: Hailu, Andreas [Engineering] ; Piotr Nowojski Cc: user@flink.apache.org Subject: Re: CoGroup SortMerger performance

Re: CoGroup SortMerger performance degradation from 1.6.4 - 1.9.1?

. -- From:Hailu, Andreas Send Time:2019 Nov. 21 (Thu.) 23:37 To:Piotr Nowojski Cc:Zhijiang ; user@flink.apache.org Subject:RE: CoGroup SortMerger performance degradation from 1.6.4 - 1.9.1? Thanks, Piotr. We’ll rerun our apps today with this and get back to you

RE: CoGroup SortMerger performance degradation from 1.6.4 - 1.9.1?

Thanks, Piotr. We’ll rerun our apps today with this and get back to you. // ah From: Piotr Nowojski On Behalf Of Piotr Nowojski Sent: Thursday, November 21, 2019 10:14 AM To: Hailu, Andreas [Engineering] Cc: Zhijiang ; user@flink.apache.org Subject: Re: CoGroup SortMerger performance

Re: CoGroup SortMerger performance degradation from 1.6.4 - 1.9.1?

<>From: Zhijiang > Sent: Wednesday, November 20, 2019 11:32 PM > To: Hailu, Andreas [Engineering] ; > user@flink.apache.org > Subject: Re: CoGroup SortMerger performance degradation from 1.6.4 - 1.9.1? > > Hi Andreas, > > You are running a batch job, so the

RE: CoGroup SortMerger performance degradation from 1.6.4 - 1.9.1?

6.65GB, so it sounds like the problem lies somewhere in the changes around mapped memory. // ah From: Zhijiang Sent: Wednesday, November 20, 2019 11:32 PM To: Hailu, Andreas [Engineering] ; user@flink.apache.org Subject: Re: CoGroup SortMerger performance degradation from 1.6.4 - 1.9.1? Hi Andreas

Re: CoGroup SortMerger performance degradation from 1.6.4 - 1.9.1?

stack, especially it really spans several releases. Best, Zhijiang -- From:Hailu, Andreas Send Time:2019 Nov. 21 (Thu.) 01:03 To:user@flink.apache.org Subject:RE: CoGroup SortMerger performance degradation from 1.6.4 - 1.9.1

RE: CoGroup SortMerger performance degradation from 1.6.4 - 1.9.1?

, 2019 6:01 PM To: 'user@flink.apache.org' Subject: CoGroup SortMerger performance degradation from 1.6.4 - 1.9.1? Hi, We're in the middle of testing the upgrade of our data processing flows from Flink 1.6.4 to 1.9.1. We're seeing that flows which were running just fine on 1.6

CoGroup SortMerger performance degradation from 1.6.4 - 1.9.1?

around how the data is sorted prior to being fed to the CoGroup operator - this is the error that we encounter: Caused by: org.apache.flink.runtime.client.JobExecutionException: Job execution failed. at org.apache.flink.runtime.jobmaster.JobResult.toJobExecutionResult(JobResult.ja

Re: RuntimeException with valve output watermark when using CoGroup

Thanks for the update Taneli. Glad that you solved the problem. If you should find out more about the more obscure case, let us know. Maybe there is something we can still improve to prevent misleading exceptions in the future. Cheers, Till On Tue, Jan 1, 2019 at 3:01 PM Taneli Saastamoinen < tan

Re: RuntimeException with valve output watermark when using CoGroup

To return to this old thing, this was basically user error. The second of the transformations was keying by a field that was sometimes null after the first transformation. (This was supposed to never happen, but then it did happen in production.) The confusing part is where the exception occurs. T

Re: RuntimeException with valve output watermark when using CoGroup

On 27 July 2018 at 19:21, Chesnay Schepler wrote: > At first glance this looks like a bug. Is the nothing in the stack trace after the NullPointerException? Hmm, there is actually, sorry about that: Caused by: java.lang.NullPointerException at org.apache.flink.runtime.state.KeyGroupRangeAssignme

Re: RuntimeException with valve output watermark when using CoGroup

e a bug in Flink. My code is too proprietary to be shared directly, but here's the general gist. I'm getting data in as JSON, parsing it into POJOs, and then aggregating those with coGroup(), taking the maximum of two separate fields. I then take the results of this and aggregate i

RuntimeException with valve output watermark when using CoGroup

rmark(StreamInputProcessor.java:265) I'm not sure why this is happening but I suspect it could be a bug in Flink. My code is too proprietary to be shared directly, but here's the general gist. I'm getting data in as JSON, parsing it into POJOs, and then aggregating those with coGroup(), t

Re: coGroup exception or something else in Gelly job

ask.java:702) at java.lang.Thread.run(Thread.java:748) I guess the trigger is the coGroup function, but I’m not sure and need your help. Best, Marc

Re: coGroup exception or something else in Gelly job

erators.BatchTask.invoke(BatchTask.java:355) at org.apache.flink.runtime.taskmanager.Task.run(Task.java:702) at java.lang.Thread.run(Thread.java:748) I guess the trigger is the coGroup function, but I’m not sure and need your help. Best, Marc

Re: coGroup exception or something else in Gelly job

tors.BatchTask.run(BatchTask.java:490) at org.apache.flink.runtime.operators.BatchTask.invoke(BatchTask.java:355) at org.apache.flink.runtime.taskmanager.Task.run(Task.java:702) at java.lang.Thread.run(Thread.java:748) I guess the trigger is the coGroup function, but I’m not sure and need your help. Best, Marc

Re: coGroup exception or something else in Gelly job

> at scala.concurrent.forkjoin.ForkJoinPool.runWorker( > ForkJoinPool.java:1979) > at scala.concurrent.forkjoin.ForkJoinWorkerThread.run( > ForkJoinWorkerThread.java:107) > Caused by: java.lang.NullPointerException > at org.apache.flink.gelly.GraphExtension.CustomVertexValue. > createInitSemiCluster(CustomVertexValue.ja

Re: coGroup exception or something else in Gelly job

run(Task.java:702) at java.lang.Thread.run(Thread.java:748) I guess the trigger is the coGroup function, but I’m not sure and need your help. Best, Marc

Re: coGroup exception or something else in Gelly job

2) > at > org.apache.flink.runtime.operators.CoGroupDriver.run(CoGroupDriver.java:159) > at org.apache.flink.runtime.operators.BatchTask.run(BatchTask.java:490) > at org.apache.flink.runtime.operators.BatchTask.invoke(BatchTask.java:355) > at org.apache.flink.runtime.taskmanager.Task.run(Task.java:702) > at java.lang.Thread.run(Thread.java:748) > > I guess the trigger is the coGroup function, but I’m not sure and need your > help. > > > Best, > > Marc >

coGroup exception or something else in Gelly job

java.lang.Thread.run(Thread.java:748) I guess the trigger is the coGroup function, but I’m not sure and need your help. Best, Marc

Re: Cogroup hints/performance

Hi Billy, A CoGroup does not have any freedom in its execution strategy. It requires that both inputs are partitioned on the grouping keys and are then performs a local sort-merge join, i.e, both inputs are sorted. Existing partitioning or sort orders can be reused. Since there is only one

Cogroup hints/performance

We have a cogroup where sometimes we cogroup like this: Dataset z = larger.coGroup(small).where... The strategy is printed as hash on key and a sort asc on the other key. Which is which? Naively, we'd want to hash larger and sort the small? Or is that wrong? What factors would impac

Re: PartitionedState and watermark of Window coGroup()

these two work when writing trigger for window > coGroup(). > > Stream1.assignTimestampsAndWatermarks(new EventWatermark()) > .coGroup(Stream2.assignTimestampsAndWatermarks(new > EventWatermark())) > .where(new JSONKey("key"

PartitionedState and watermark of Window coGroup()

Hi Flink users, I'm a bit confused about how these two work when writing trigger for window coGroup(). Stream1.assignTimestampsAndWatermarks(new EventWatermark()) .coGroup(Stream2.assignTimestampsAndWatermarks(new EventWatermark())) .where(new JSONKey

Re: Re: 回复：Re: modify coGroup GlobalWindows_GlobalWindow

uot; have been > processed, the right side of your coGroup will always be empty no matter > what is incoming in your socketTextStream.",the mean i can not get. > the following is the ideal from me(it maybe error): > the coGroup will create new dataStream,T1 and T2,this must use >

回复：Re: 回复：Re: modify coGroup GlobalWindows_GlobalWindow

think your anwser. but i can not get your ideal."If all elements of "words2" have been processed, the right side of your coGroup will always be empty no matter what is incoming in your socketTextStream.",the mean i can not get. the following is the ideal from me(it maybe

Re: 回复：Re: modify coGroup GlobalWindows GlobalWindow

I think you have to rethink your approach. In your example "words2" is a stream but only with a fixed set of elements. If all elements of "words2" have been processed, the right side of your coGroup will always be empty no matter what is incoming in your socketTextStream

回复：Re: modify coGroup GlobalWindows GlobalWindow

i try read data into a list or List[Map] to store the T2,but i think if use list or List[Map],there is not parallelization,so i want to use coGroup. other hand,the coGroup function is join the T1 and T2,and must have window and trigger method,the window is cut the T1 and T2, the trigger is

Re: modify coGroup GlobalWindows GlobalWindow

Hi, will words2 always remain constant? If yes, you don't have to create a stream out of it and coGroup it, but you could simply pass the collection to Map/FlatMap function and do the joining there without the need of a window. Btw. you know that non-keyed global windows do not scale?

modify coGroup GlobalWindows GlobalWindow

t;),("c","w3"),("d","w4")) val joinedStream = words1 .coGroup(words2) .where(_._1) .equalTo(_._1) .window(GlobalWindows.create()) .trigger(CountTrigger.of(1)) val res = joinedStream.apply(new InnerJoinFunction).print() env.exe

Re: Hash tables - joins, cogroup, deltaIteration

rouping/aggregations)?* > > I am interested in the pending work progress and also if you consider to > add an implementation where Joins and Solution Set in delta iterations (and > CoGroup) can rely on a hybrid implementation where the engine can use also > disk if not enough memor

Hash tables - joins, cogroup, deltaIteration

if you consider to add an implementation where Joins and Solution Set in delta iterations (and CoGroup) can rely on a hybrid implementation where the engine can use also disk if not enough memory available when working with these hash tables. [1] https://cwiki.apache.org/confluence/pages

Re: Different CoGroup behavior inside DeltaIteration

Hi, Thanks for the suggestion. I'm trying to use the delta iteration so that I can get the empty work set convergence criteria for free. But since doing an outer join between the work set and the solution set is not possible using cogroup, I will try to adapt my algorithm to use the

Re: Different CoGroup behavior inside DeltaIteration

It is actually very important that the co group in delta iterations works like that. If the CoGroup touched every element in the solution set, the "decreasing work" effect would not happen. The delta iterations are designed for cases where specific updates to the solution are made, dri

Re: Different CoGroup behavior inside DeltaIteration

Hi, this is an artifact of how the solution set is internally implemented. Usually, a CoGroup is executed using a sort-merge strategy, i.e., both input are sorted, merged, and handed to the CoGroup function in a streaming fashion. Both inputs are treated equally, and if one of both inputs does

Different CoGroup behavior inside DeltaIteration

Hi, When running CoGroup between the solution set and a different dataset inside a DeltaIteration, the CoGroupFunction only get called for items that exist in the other dataset, simillar to an inner join. This is not the documented behavior for CoGroup: If a DataSet has a group with no

Re: what different between join and coGroup in flink

Yes. On 09/06/2015 10:55 PM, hagersaleh wrote: > Join can be executed more efficiently than CoGroup > this means Join faster than COGroup in executed > > > > > -- > View this message in context: > http://apache-flink-user-mailing-list-archive.2336050.n4.nabble

Re: what different between join and coGroup in flink

CoGroup is more generic than Join. You can perform a Join with CoGroup but not do a CoGroup with a Join. However, Join can be executed more efficiently than CoGroup. 2015-09-03 22:28 GMT+02:00 hagersaleh : > what different between join and coGroup in flink > > > > > -- >

Re: sorted cogroup

Hi Michele, Flink supports coGroups on sorted inputs. If you have a ds1 = DataSet[(Key, Value1)] and ds2 = DataSet[(Key, Value2)] you obtain a sorted coGroup for example by: ds1.coGroup(ds2).where(0).equalsTo(0).sortFirstGroup(1, Order.ASCENDING).sortSecondGroup(1, Order.DESCENDING) Cheers

sorted cogroup

Hi everybody, i need to execute a cogroup on sorted groups. I explain it better: I have two datasets i.e. (key, value), I want to cogroup on key and then the have both iterator sorted by value how can i get it? I know iterator should be collected to be sorted but i want to avoid it. what happens

Re: cogroup

ok thanks! then by now i will use it until true outer join is ready Il giorno 29/giu/2015, alle ore 18:22, Fabian Hueske mailto:fhue...@gmail.com>> ha scritto: Yes, if you need outer join semantics you have to go with CoGroup. Some members of the Flink community are working on true outer

Re: cogroup

Yes, if you need outer join semantics you have to go with CoGroup. Some members of the Flink community are working on true outer joins for Flink, but I don't know what the progress is. Best, Fabian 2015-06-29 18:05 GMT+02:00 Michele Bertoni : > thanks both for answering, > th

Re: cogroup

thanks both for answering, that’s what i expected I was using join at first but sadly i had to move from join to cogroup because I need outer join the alternative to the cogroup is to “complete” the inner join extracting from the original dataset what did not matched in the cogroup by

Re: cogroup

If you just want to do the pairwise comparison try join(). Join is an inner join and will give you all pairs of elements with matching keys. For CoGroup, there is no other way than collecting one side in memory. Best, Fabian 2015-06-29 17:42 GMT+02:00 Matthias J. Sax : > Why do you not us

Re: cogroup

Why do you not use a join? CoGroup seems not to be the right operator. -Matthias On 06/29/2015 05:40 PM, Michele Bertoni wrote: > Hi I have a question on cogroup > > when I cogroup two dataset is there a way to compare each element on the left > with each element on the right (in

cogroup

Hi I have a question on cogroup when I cogroup two dataset is there a way to compare each element on the left with each element on the right (inside a group) without collecting one side? right now I am doing left.cogroup(right).where(0,1,2).equalTo(0,1,2){ (leftIterator, rightIterator

Re: coGroup Iterator NoSuchElement

>> >> How to fix that ?!! >> >> On Thu, Jun 4, 2015 at 11:00 AM, Fabian Hueske wrote: >> >>> Hi, >>> >>> one of the iterables of a CoGroup function can be empty. Calling >>> iterator.next() on an empty iterator raises the NoSuchElemen

Re: coGroup Iterator NoSuchElement

common key, so it will be normal to have many tuples on > side, which does not exist on the other side .. > > How to fix that ?!! > > On Thu, Jun 4, 2015 at 11:00 AM, Fabian Hueske wrote: > >> Hi, >> >> one of the iterables of a CoGroup function can be empty. Calli

Re: coGroup Iterator NoSuchElement

The regular JOIN has the semantics of an inner join, filtering out cases where no matching tuple is found on one side. CoGroup follows the semantics of an outer join on groups, delivering also empty groups on some sides. On Thu, Jun 4, 2015 at 11:18 AM, Mustafa Elbehery wrote: > Yes,

Re: coGroup Iterator NoSuchElement

ne of the iterables of a CoGroup function can be empty. Calling > iterator.next() on an empty iterator raises the NoSuchElementException. > This is the expected behavior of the function. > > Are you sure your assumption about your data are correct, i.e., that the > iterator should al

Re: coGroup Iterator NoSuchElement

Hi, one of the iterables of a CoGroup function can be empty. Calling iterator.next() on an empty iterator raises the NoSuchElementException. This is the expected behavior of the function. Are you sure your assumption about your data are correct, i.e., that the iterator should always have (at

Re: coGroup Iterator NoSuchElement

Hi, public static class ComputeStudiesProfile implements CoGroupFunction { Person person; @Override public void coGroup(Iterable iterable, Iterable iterable1, Collector collector) throws Exception { Iterator iterator = iterable.iterator(); person = iterator.next

Re: coGroup Iterator NoSuchElement

.with(new ComputeJobsProfile()); > > updatedPersonTwo.print(); > > > On Wed, Jun 3, 2015 at 3:45 PM, Mustafa Elbehery < > elbeherymust...@gmail.com> wrote: > >> Hi, >> >> I am trying to write two coGrouprs in sequence on the same ETL .. In use >>

Re: coGroup Iterator NoSuchElement

.where("name").equalTo("name") .with(new ComputeJobsProfile()); updatedPersonTwo.print(); On Wed, Jun 3, 2015 at 3:45 PM, Mustafa Elbehery wrote: > Hi, > > I am trying to write two coGrouprs in sequence on the same ETL .. In use > co

coGroup Iterator NoSuchElement

Hi, I am trying to write two coGrouprs in sequence on the same ETL .. In use common dataset in both of them, in the first coGroup I update the initial dataset and retrieve the result in a new dataset object. Then I use the result in the second coGroup with another new dataset. While debugging, I

Re: using BroadcastSet in Join/CoGroup/Cross

text: > http://apache-flink-incubator-user-mailing-list-archive.2336050.n4.nabble.com/using-BroadcastSet-in-Join-CoGroup-Cross-tp864.html > Sent from the Apache Flink (Incubator) User Mailing List archive. mailing > list archive at Nabble.com. >

using BroadcastSet in Join/CoGroup/Cross

hello, Is it possible to use .withBroadcastSet in other operations than Map, says Join for example? -- View this message in context: http://apache-flink-incubator-user-mailing-list-archive.2336050.n4.nabble.com/using-BroadcastSet-in-Join-CoGroup-Cross-tp864.html Sent from the Apache Flink