kafka streams: join 2 Windowed tables

[Ara Ebrahimi <ar...@argyledata.com>]

Hi,

Is joining 2 streams based on Windowed<String> keys supposed to work?

I have 2 KTables:
- KTable<Windowed<String>, T> events: I process events and aggregate events that have a common criteria using aggregateByKey and UnlimitedWindows as window (for now)
- KTable<Windowed<String>, S> hourlyStats: I calculate some stats using aggregateByKey for hourly windows TimeWindows.of(“window name”, hourly)

Since both use aggregateByKey() they are both KTables and both have Windowed<String> keys.

I need to leftJoin the first one (events) with the second one (hourlyStats) BUT I need to join event x which occurred at time t0 with hourlyStats of the t0 window. In other words I need to join using JoinWindows.of("JoinWindow").within(60 * 60 * 1000). Since both are KTables this is not possible. But if I turn them both into KStream's using toStream() then I can use the leftJoin() variants which supports JoinWindows. They would both be KStream<Windowed<String>, ...>.

The problem is the join doesn’t really happen. No hourlyStats is actually found for any row of events. The TimestampExtractor for both is correct.

So, Is joining 2 streams based on Windowed<String> keys supposed to work? If not then how can I accomplish the above task?

Thanks,
Ara.



________________________________

This message is for the designated recipient only and may contain privileged, proprietary, or otherwise confidential information. If you have received it in error, please notify the sender immediately and delete the original. Any other use of the e-mail by you is prohibited. Thank you in advance for your cooperation.

________________________________

Re: kafka streams: join 2 Windowed tables

["Matthias J. Sax" <ma...@confluent.io>]

Hi Ara,

The problem is, that type Window<String> is a composite key of
"String-Long" that encoded your original key of type String the window
start time of type Long. Thus, even if two record have the same
"string-key" they might be different, because the belong to different
windows.

Try to insert a map() before your join that converts <Window<String>,
Value> into <String,Value>.

-Matthias


On 09/02/2016 12:30 AM, Ara Ebrahimi wrote:
> Hi,
> 
> Is joining 2 streams based on Windowed<String> keys supposed to work?
> 
> I have 2 KTables:
> - KTable<Windowed<String>, T> events: I process events and aggregate events that have a common criteria using aggregateByKey and UnlimitedWindows as window (for now)
> - KTable<Windowed<String>, S> hourlyStats: I calculate some stats using aggregateByKey for hourly windows TimeWindows.of(“window name”, hourly)
> 
> Since both use aggregateByKey() they are both KTables and both have Windowed<String> keys.
> 
> I need to leftJoin the first one (events) with the second one (hourlyStats) BUT I need to join event x which occurred at time t0 with hourlyStats of the t0 window. In other words I need to join using JoinWindows.of("JoinWindow").within(60 * 60 * 1000). Since both are KTables this is not possible. But if I turn them both into KStream's using toStream() then I can use the leftJoin() variants which supports JoinWindows. They would both be KStream<Windowed<String>, ...>.
> 
> The problem is the join doesn’t really happen. No hourlyStats is actually found for any row of events. The TimestampExtractor for both is correct.
> 
> So, Is joining 2 streams based on Windowed<String> keys supposed to work? If not then how can I accomplish the above task?
> 
> Thanks,
> Ara.
> 
> 
> 
> ________________________________
> 
> This message is for the designated recipient only and may contain privileged, proprietary, or otherwise confidential information. If you have received it in error, please notify the sender immediately and delete the original. Any other use of the e-mail by you is prohibited. Thank you in advance for your cooperation.
> 
> ________________________________
>