Because I want to group them for the last X minutes. In this case last 1
minute.
On Mon, Feb 14, 2022 at 10:01 AM Ali Bahadir Zeybek <a...@ververica.com>
wrote:
> Hello John,
>
> Then may I ask you why you need to use a time attribute as part of your
> key?
> Why not just key by the fields like `mydomain.com` and `some-article` in
> your
> example and use only window operator for grouping elements based on time?
>
> Sincerely,
>
> Ali
>
> On Mon, Feb 14, 2022 at 3:55 PM John Smith <java.dev....@gmail.com> wrote:
>
>> Hi, thanks. As previously mentioned, processing time. So I
>> regardless when the event was generated I want to count all events I have
>> right now (as soon as they are seen by the flink job).
>>
>> On Mon, Feb 14, 2022 at 4:16 AM Ali Bahadir Zeybek <a...@ververica.com>
>> wrote:
>>
>>> Hello John,
>>>
>>> Currently you are grouping the elements two times based on some time
>>> attribute, one while keying - with event time - and one while windowing
>>> - with
>>> processing time. Therefore, the windowing mechanism produces a new window
>>> computation when you see an element with the same key but arrived later
>>> from
>>> the previous window start and end timestamps. Can you please clarify with
>>> which notion of time you would like to handle the stream of data?
>>>
>>> Sincerely,
>>>
>>> Ali
>>>
>>> On Fri, Feb 11, 2022 at 6:43 PM John Smith <java.dev....@gmail.com>
>>> wrote:
>>>
>>>> Ok I used the method suggested by Ali. The error is gone. But now I see
>>>> multiple counts emitted for the same key...
>>>>
>>>> DataStream<MyEvent> slStream = env.fromSource(kafkaSource,
>>>> WatermarkStrategy.noWatermarks(), "Kafka Source")
>>>> .uid(kafkaTopic).name(kafkaTopic)
>>>> .setParallelism(kafkaParallelism)
>>>> .flatMap(new MapToMyEvent("my-event", windowSizeMins, "message"))
>>>> <------ Timestamp in GMT created here rounded to the closest minute down.
>>>> .uid("map-json-logs").name("map-json-logs");
>>>>
>>>> slStream.keyBy(new MinutesKeySelector())
>>>>
>>>> .window(TumblingProcessingTimeWindows.of(Time.minutes(windowSizeMins)))
>>>> <---- Tumbling window of 1 minute.
>>>>
>>>>
>>>>
>>>> So below you will see a new count was emitted at 16:51 and 16:55
>>>>
>>>> {"countId":"2022-02-11T16:50:00Z|mydomain.com
>>>> |/some-article","countDateTime":"2022-02-11T16:50:00Z","domain":"
>>>> mydomain.com","uri":"/some-article","count":3542}
>>>> -----
>>>> {"countId":"2022-02-11T16:51:00Z|mydomain.com
>>>> |/some-article","countDateTime":"2022-02-11T16:51:00Z","domain":"
>>>> mydomain.com","uri":"/some-article","count":16503}
>>>> {"countId":"2022-02-11T16:51:00Z|mydomain.com
>>>> |/some-article","countDateTime":"2022-02-11T16:51:00Z","domain":"
>>>> mydomain.com","uri":"/some-article","count":70}
>>>> -----
>>>>
>>>> {"countId":"2022-02-11T16:52:00Z|mydomain.com
>>>> |/some-article","countDateTime":"2022-02-11T16:52:00Z","domain":"
>>>> mydomain.com","uri":"/some-article","count":16037}
>>>> {"countId":"2022-02-11T16:53:00Z|mydomain.com
>>>> |/some-article","countDateTime":"2022-02-11T16:53:00Z","domain":"
>>>> mydomain.com","uri":"/some-article","count":18679}
>>>> {"countId":"2022-02-11T16:54:00Z|mydomain.com
>>>> |/some-article","countDateTime":"2022-02-11T16:54:00Z","domain":"
>>>> mydomain.com","uri":"/some-article","count":17697}
>>>> -----
>>>>
>>>> {"countId":"2022-02-11T16:55:00Z|mydomain.com
>>>> |/some-article","countDateTime":"2022-02-11T16:55:00Z","domain":"
>>>> mydomain.com","uri":"/some-article","count":18066}
>>>> {"countId":"2022-02-11T16:55:00Z|mydomain.com
>>>> |/some-article","countDateTime":"2022-02-11T16:55:00Z","domain":"
>>>> mydomain.com","uri":"/some-article","count":58}
>>>> -----
>>>> {"countId":"2022-02-11T16:56:00Z|mydomain.com
>>>> |/some-article","countDateTime":"2022-02-11T16:56:00Z","domain":"
>>>> mydomain.com","uri":"/some-article","count":17489}
>>>>
>>>>
>>>>
>>>>
>>>> On Mon, Feb 7, 2022 at 12:44 PM John Smith <java.dev....@gmail.com>
>>>> wrote:
>>>>
>>>>> Ok I think Ali's solution makes the most sense to me. I'll try it and
>>>>> let you know.
>>>>>
>>>>> On Mon, Feb 7, 2022 at 11:44 AM Jing Ge <j...@ververica.com> wrote:
>>>>>
>>>>>> Hi John,
>>>>>>
>>>>>> your getKey() implementation shows that it is not deterministic,
>>>>>> since calling it with the same click instance multiple times will return
>>>>>> different keys. For example a call at 12:01:59.950 and a call at
>>>>>> 12:02:00.050 with the same click instance will return two different keys:
>>>>>>
>>>>>> 2022-04-07T12:01:00.000Z|cnn.com|some-article-name
>>>>>> 2022-04-07T12:02:00.000Z|cnn.com|some-article-name
>>>>>>
>>>>>> best regards
>>>>>> Jing
>>>>>>
>>>>>> On Mon, Feb 7, 2022 at 5:07 PM John Smith <java.dev....@gmail.com>
>>>>>> wrote:
>>>>>>
>>>>>>> Maybe there's a misunderstanding. But basically I want to
>>>>>>> do clickstream count for a given "url" and for simplicity and accuracy
>>>>>>> of
>>>>>>> the count base it on processing time (event time doesn't matter as long
>>>>>>> as
>>>>>>> I get a total of clicks at that given processing time)
>>>>>>>
>>>>>>> So regardless of the event time. I want all clicks for the current
>>>>>>> processing time rounded to the minute per link.
>>>>>>>
>>>>>>> So, if now was 2022-04-07T12:01:00.000Z
>>>>>>>
>>>>>>> Then I would want the following result...
>>>>>>>
>>>>>>> 2022-04-07T12:01:00.000Z|cnn.com|some-article-name count = 10
>>>>>>> 2022-04-07T12:01:00.000Z|cnn.com|some-other-article count = 2
>>>>>>> 2022-04-07T12:01:00.000Z|cnn.com|another-article count = 15
>>>>>>> ....
>>>>>>> 2022-04-07T12:02:00.000Z|cnn.com|some-article-name count = 30
>>>>>>> 2022-04-07T12:02:00.000Z|cnn.com|some-other-article count = 1
>>>>>>> 2022-04-07T12:02:00.000Z|cnn.com|another-article count = 10
>>>>>>> And so on...
>>>>>>>
>>>>>>> @Override
>>>>>>> public MyEventCountKey getKey(final MyEvent click) throws Exception
>>>>>>> {
>>>>>>> MyEventCountKey key = new MyEventCountKey(
>>>>>>> Instant.from(roundFloor(Instant.now().atZone(ZoneId.of("UTC")),
>>>>>>> ChronoField.MINUTE_OF_HOUR, windowSizeMins)).toString(),
>>>>>>> click.getDomain(), // cnn.com
>>>>>>> click.getPath(), // /some-article-name
>>>>>>> );
>>>>>>> return key;
>>>>>>> }
>>>>>>>
>>>>>>>
>>>>>>>
>>>>>>> On Mon, Feb 7, 2022 at 10:48 AM David Morávek <d...@apache.org>
>>>>>>> wrote:
>>>>>>>
>>>>>>>> The key selector works.
>>>>>>>>
>>>>>>>>
>>>>>>>> No it does not ;) It depends on the system time so it's not
>>>>>>>> deterministic (you can get different keys for the very same element).
>>>>>>>>
>>>>>>>> How do you key a count based on the time. I have taken this from
>>>>>>>>> samples online.
>>>>>>>>>
>>>>>>>>
>>>>>>>> This is what the windowing is for. You basically want to group /
>>>>>>>> combine elements per key and event time window [1].
>>>>>>>>
>>>>>>>> [1]
>>>>>>>> https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/dev/datastream/operators/windows/
>>>>>>>>
>>>>>>>> Best,
>>>>>>>> D.
>>>>>>>>
>>>>>>>> On Mon, Feb 7, 2022 at 3:44 PM John Smith <java.dev....@gmail.com>
>>>>>>>> wrote:
>>>>>>>>
>>>>>>>>> The key selector works. It only causes an issue if there too many
>>>>>>>>> keys produced in one shot. For example of 100 "same" keys are
>>>>>>>>> produced for
>>>>>>>>> that 1 minutes it's ok. But if 101 are produced the error happens.
>>>>>>>>>
>>>>>>>>>
>>>>>>>>> If you look at the reproducer at least that's what's hapenning
>>>>>>>>>
>>>>>>>>> How do you key a count based on the time. I have taken this from
>>>>>>>>> samples online.
>>>>>>>>>
>>>>>>>>> The key is that particular time for that particular URL path.
>>>>>>>>>
>>>>>>>>> So cnn.com/article1 was clicked 10 times at 2022-01-01T10:01:00
>>>>>>>>>
>>>>>>>>> On Mon., Feb. 7, 2022, 8:57 a.m. Chesnay Schepler, <
>>>>>>>>> ches...@apache.org> wrote:
>>>>>>>>>
>>>>>>>>>> Your Key selector doesn't need to implement hashCode, but given
>>>>>>>>>> the same object it has to return the same key.
>>>>>>>>>> In your reproducer the returned key will have different
>>>>>>>>>> timestamps, and since the timestamp is included in the hashCode,
>>>>>>>>>> they will
>>>>>>>>>> be different each time.
>>>>>>>>>>
>>>>>>>>>> On 07/02/2022 14:50, John Smith wrote:
>>>>>>>>>>
>>>>>>>>>> I don't get it? I provided the reproducer. I implemented the
>>>>>>>>>> interface to Key selector it needs hashcode and equals as well?
>>>>>>>>>>
>>>>>>>>>> I'm attempting to do click stream. So the key is based on
>>>>>>>>>> processing date/time rounded to the minute + domain name + path
>>>>>>>>>>
>>>>>>>>>> So these should be valid below?
>>>>>>>>>>
>>>>>>>>>> 2022-01-01T10:02:00 + cnn.com + /article1
>>>>>>>>>> 2022-01-01T10:02:00 + cnn.com + /article1
>>>>>>>>>> 2022-01-01T10:02:00 + cnn.com + /article1
>>>>>>>>>>
>>>>>>>>>> 2022-01-01T10:02:00 + cnn.com + /article2
>>>>>>>>>>
>>>>>>>>>> 2022-01-01T10:03:00 + cnn.com + /article1
>>>>>>>>>> 2022-01-01T10:03:00 + cnn.com + /article1
>>>>>>>>>>
>>>>>>>>>> 2022-01-01T10:03:00 + cnn.com + /article3
>>>>>>>>>> 2022-01-01T10:03:00 + cnn.com + /article3
>>>>>>>>>>
>>>>>>>>>> On Mon., Feb. 7, 2022, 2:53 a.m. Chesnay Schepler, <
>>>>>>>>>> ches...@apache.org> wrote:
>>>>>>>>>>
>>>>>>>>>>> Don't KeySelectors also need to be deterministic?
>>>>>>>>>>>
>>>>>>>>>>> * The {@link KeySelector} allows to use deterministic objects for
>>>>>>>>>>> operations such as reduce,* reduceGroup, join, coGroup, etc. *If
>>>>>>>>>>> invoked multiple times on the same object, the returned key*** must
>>>>>>>>>>> be the same.*
>>>>>>>>>>>
>>>>>>>>>>>
>>>>>>>>>>> On 04/02/2022 18:25, John Smith wrote:
>>>>>>>>>>>
>>>>>>>>>>> Hi Francesco, here is the reproducer:
>>>>>>>>>>> https://github.com/javadevmtl/flink-key-reproducer
>>>>>>>>>>>
>>>>>>>>>>> So, essentially it looks like when there's a high influx of
>>>>>>>>>>> records produced from the source that the Exception is thrown.
>>>>>>>>>>>
>>>>>>>>>>> The key is generated by 3 values: date/time rounded to the
>>>>>>>>>>> minute and 2 strings.
>>>>>>>>>>> So you will see keys as follows...
>>>>>>>>>>> 2022-02-04T17:20:00Z|foo|bar
>>>>>>>>>>> 2022-02-04T17:21:00Z|foo|bar
>>>>>>>>>>> 2022-02-04T17:22:00Z|foo|bar
>>>>>>>>>>>
>>>>>>>>>>> The reproducer has a custom source that basically produces a
>>>>>>>>>>> record in a loop and sleeps for a specified period of milliseconds
>>>>>>>>>>> 100ms in
>>>>>>>>>>> this case.
>>>>>>>>>>> The lower the sleep delay the faster records are produced the
>>>>>>>>>>> more chances the exception is thrown. With a 100ms delay it's always
>>>>>>>>>>> thrown. Setting a 2000 to 3000ms will guarantee it to work.
>>>>>>>>>>> The original job uses a Kafka Source so it should technically be
>>>>>>>>>>> able to handle even a couple thousand records per second.
>>>>>>>>>>>
>>>>>>>>>>>
>>>>>>>>>>> On Thu, 3 Feb 2022 at 16:41, John Smith <java.dev....@gmail.com>
>>>>>>>>>>> wrote:
>>>>>>>>>>>
>>>>>>>>>>>> Ok it's not my data either. I think it may be a volume issue. I
>>>>>>>>>>>> have managed to consistently reproduce the error. I'll upload a
>>>>>>>>>>>> reproducer
>>>>>>>>>>>> ASAP.
>>>>>>>>>>>>
>>>>>>>>>>>>
>>>>>>>>>>>>
>>>>>>>>>>>> On Thu, 3 Feb 2022 at 15:37, John Smith <java.dev....@gmail.com>
>>>>>>>>>>>> wrote:
>>>>>>>>>>>>
>>>>>>>>>>>>> Ok so I tried to create a reproducer but I couldn't reproduce
>>>>>>>>>>>>> it. But the actual job once in a while throws that error. So I'm
>>>>>>>>>>>>> wondering
>>>>>>>>>>>>> if maybe one of the records that comes in is not valid, though I
>>>>>>>>>>>>> do
>>>>>>>>>>>>> validate prior to getting to the key and window operators.
>>>>>>>>>>>>>
>>>>>>>>>>>>> On Thu, 3 Feb 2022 at 14:32, John Smith <
>>>>>>>>>>>>> java.dev....@gmail.com> wrote:
>>>>>>>>>>>>>
>>>>>>>>>>>>>> Actually maybe not because with PrintSinkFunction it ran for
>>>>>>>>>>>>>> a bit and then it threw the error.
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> On Thu, 3 Feb 2022 at 14:24, John Smith <
>>>>>>>>>>>>>> java.dev....@gmail.com> wrote:
>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> Ok it may be the ElasticSearch connector causing the issue?
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> If I use PrintSinkFunction then I get no error and my stats
>>>>>>>>>>>>>>> print as expected.
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> On Wed, 2 Feb 2022 at 03:01, Francesco Guardiani <
>>>>>>>>>>>>>>> france...@ververica.com> wrote:
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> Hi,
>>>>>>>>>>>>>>>> your hash code and equals seems correct. Can you post a
>>>>>>>>>>>>>>>> minimum stream pipeline reproducer using this class?
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> FG
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> On Tue, Feb 1, 2022 at 8:39 PM John Smith <
>>>>>>>>>>>>>>>> java.dev....@gmail.com> wrote:
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> Hi, getting java.lang.IllegalArgumentException: Key group
>>>>>>>>>>>>>>>>> 39 is not in KeyGroupRange{startKeyGroup=96,
>>>>>>>>>>>>>>>>> endKeyGroup=103}. Unless
>>>>>>>>>>>>>>>>> you're directly using low level state access APIs, this is
>>>>>>>>>>>>>>>>> most likely
>>>>>>>>>>>>>>>>> caused by non-deterministic shuffle key (hashCode and equals
>>>>>>>>>>>>>>>>> implementation).
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> This is my class, is my hashCode deterministic?
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> public final class MyEventCountKey {
>>>>>>>>>>>>>>>>> private final String countDateTime; private final
>>>>>>>>>>>>>>>>> String domain; private final String event; public
>>>>>>>>>>>>>>>>> MyEventCountKey(final String countDateTime, final String
>>>>>>>>>>>>>>>>> domain, final String event) {
>>>>>>>>>>>>>>>>> this.countDateTime = countDateTime;
>>>>>>>>>>>>>>>>> this.domain = domain; this.event = event; }
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> public String getCountDateTime() {
>>>>>>>>>>>>>>>>> return countDateTime; }
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> public String getDomain() {
>>>>>>>>>>>>>>>>> return domain; }
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> public String getEven() {
>>>>>>>>>>>>>>>>> return event; }
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> @Override public String toString() {
>>>>>>>>>>>>>>>>> return countDateTime + "|" + domain + "|" + event;
>>>>>>>>>>>>>>>>> }
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> @Override public boolean equals(Object o) {
>>>>>>>>>>>>>>>>> if (this == o) return true; if (o == null ||
>>>>>>>>>>>>>>>>> getClass() != o.getClass()) return false;
>>>>>>>>>>>>>>>>> MyEventCountKey that = (MyEventCountKey) o; return
>>>>>>>>>>>>>>>>> countDateTime.equals(that.countDateTime) &&
>>>>>>>>>>>>>>>>> domain.equals(that.domain) &&
>>>>>>>>>>>>>>>>> event.equals(that.event); }
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> @Override public int hashCode() {
>>>>>>>>>>>>>>>>> final int prime = 31; int result = 1;
>>>>>>>>>>>>>>>>> result = prime * result + countDateTime.hashCode();
>>>>>>>>>>>>>>>>> result = prime * result + domain.hashCode(); result =
>>>>>>>>>>>>>>>>> prime * result + event.hashCode(); return result; }
>>>>>>>>>>>>>>>>> }
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>
>>>>>>>>>>
