Explicit Keyed Streams

[Max-Meldrum]

Currently key(s) are declared through the Arcon derive macro that makes sure the get_key() function is implemented.

#[derive(Arcon, Arrow, prost::Message, Copy, Clone)]
#[arcon(unsafe_ser_id = 12, reliable_ser_id = 13, version = 1, keys = "id")]
pub struct Event {
    #[prost(uint64)]
    pub id: u64,
    #[prost(float)]
    pub data: f32,
}

This is however bit confusing when working with the API. Would be better to have it more explicit and introduce a KeyExtractor that the runtime can use (see below).

let mut app = Application::with_conf(conf)
    .iterator((0..1000000).map(|x| Event { id: x, data: 1.5 }), |conf| {
        conf.set_timestamp_extractor(|x: &Event| x.id);
    })
    .key_by(|event| event.id) // KeyExtractor 
     // KeyedStream?
    .operator(...)

[segeljakt]

It looks nice, I think a benefit with this is that you can have multiple key extractors for the same event type. Is the type of the stream produced by iterator also a keyed stream or something else?

[Max-Meldrum]

It looks nice, I think a benefit with this is that you can have multiple key extractors for the same event type. Is the type of the stream produced by iterator it also a keyed stream?

I take it would just be a Stream. iterator() -> Stream -> key_by(..) -> KeyedStream

We just have Stream right now, I guess we have to introduce more stream types such as in Flink Stream / KeyedStream / WindowStream?

[segeljakt]

Flink seems to have:

• DataStream<T>
• KeyedStream<T, K> extends DataStream<T>
• AllWindowedStream<T, W extends Window>
• WindowedStream<T, K, W extends Window>
• BroadcastStream<T>
• CoGroupedStreams<T1, T2>
• ConnectedStreams<IN1, IN2>
• IterativeStream<T>
• JoinedStreams<T1, T2>
• PatternStream<T>
• QueryableStateStream<K, V>

[Max-Meldrum]

One possible way of defining key_by. Use Builder notion as in other parts of the API and require an extractor + description of how the stream is keyed.

let mut app = Application::with_conf(conf)
    .iterator((0..1000000).map(|x| Event { id: x, data: 1.5 }), |conf| {
        conf.set_timestamp_extractor(|x: &Event| x.id);
    }) // Stream<Event>
    .key_by(KeyBuilder {
         extractor: Arc::new(|event| event.id),
         description: "User Name ID",
     }) // KeyedStream<Event>