Need suggestion on Flink-Kafka stream processing design

[hemant singh]

Hi,

I have different events from a device which constitutes different metrics for same device. Each of these event is produced by the device in interval of few milli seconds to a minute.

Event1(Device1) -> Stream1 -> Metric 1

Event2 (Device1) -> Stream2 -> Metric 2 ... 

..............

.......

Event100(Device1) -> Stream100 -> Metric100

The number of events can go up to few 100s for each data protocol and we have around 4-5 data protocols. Metrics from different streams makes up a records 

like for example from above example for device 1 -

Device1 -> Metric1, Metric 2, Metric15 forms a single record for the device. Currently in development phase I am using interval join to achieve this, that is to create a record with latest data from different streams(events).

Based on data protocol I have 4-5 topics. Currently the data for a single event is being pushed to a partition of the kafka topic(producer key -> event_type + data_protocol). So essentially one topic is made up of many streams. I am filtering on the key to define the streams. 

My question is - Is this correct way to stream the data, I had thought of maintaining different topic for an event, however in that case number of topics could go to few thousands and that is something which becomes little challenging to maintain and not sure if kafka handles that well.

I know there are traditional ways to do this like pushing it to timeseries db and then joining data for different metric but that is something which will never scale, also this processing should be as realtime as possible.

Are there better ways to handle this use case or I am on correct path.

Thanks,

Hemant

[r_khachatryan]

Hi Hemant,

As I understand, each protocol has a distinct set of event types (where event type == metrics type); and a distinct set of devices. Is this correct?

> Based on data protocol I have 4-5 topics. Currently the data for a single event is being pushed to a partition of the kafka topic(producer key -> event_type + data_protocol).

Here you are talking about the source (to Flink job), right?

Can you also share how are you going to consume these data?

On Mon, May 11, 2020 at 8:57 PM hemant singh <[hidden email]> wrote:

Hi,

I have different events from a device which constitutes different metrics for same device. Each of these event is produced by the device in interval of few milli seconds to a minute.

Event1(Device1) -> Stream1 -> Metric 1

Event2 (Device1) -> Stream2 -> Metric 2 ... 

..............

.......

Event100(Device1) -> Stream100 -> Metric100

The number of events can go up to few 100s for each data protocol and we have around 4-5 data protocols. Metrics from different streams makes up a records 

like for example from above example for device 1 -

Device1 -> Metric1, Metric 2, Metric15 forms a single record for the device. Currently in development phase I am using interval join to achieve this, that is to create a record with latest data from different streams(events).

Based on data protocol I have 4-5 topics. Currently the data for a single event is being pushed to a partition of the kafka topic(producer key -> event_type + data_protocol). So essentially one topic is made up of many streams. I am filtering on the key to define the streams. 

My question is - Is this correct way to stream the data, I had thought of maintaining different topic for an event, however in that case number of topics could go to few thousands and that is something which becomes little challenging to maintain and not sure if kafka handles that well.

I know there are traditional ways to do this like pushing it to timeseries db and then joining data for different metric but that is something which will never scale, also this processing should be as realtime as possible.

Are there better ways to handle this use case or I am on correct path.

Thanks,

Hemant

[hemant singh]

Hello Roman,

PFB my response -

As I understand, each protocol has a distinct set of event types (where event type == metrics type); and a distinct set of devices. Is this correct?

Yes, correct. distinct events and devices. Each device emits these event.

> Based on data protocol I have 4-5 topics. Currently the data for a single event is being pushed to a partition of the kafka topic(producer key -> event_type + data_protocol).

Here you are talking about the source (to Flink job), right?

Yes, you are right.

Can you also share how are you going to consume these data?

By consume do you mean the downstream system? 

If yes then this data will be written to a DB, some metrics goes to TSDB(Influx) as well. 

Thanks,

Hemant

On Tue, May 12, 2020 at 2:08 AM Khachatryan Roman <[hidden email]> wrote:

Hi Hemant,

As I understand, each protocol has a distinct set of event types (where event type == metrics type); and a distinct set of devices. Is this correct?

> Based on data protocol I have 4-5 topics. Currently the data for a single event is being pushed to a partition of the kafka topic(producer key -> event_type + data_protocol).

Here you are talking about the source (to Flink job), right?

Can you also share how are you going to consume these data?

Regards,
Roman

On Mon, May 11, 2020 at 8:57 PM hemant singh <[hidden email]> wrote:

Hi,

I have different events from a device which constitutes different metrics for same device. Each of these event is produced by the device in interval of few milli seconds to a minute.

Event1(Device1) -> Stream1 -> Metric 1

Event2 (Device1) -> Stream2 -> Metric 2 ... 

..............

.......

Event100(Device1) -> Stream100 -> Metric100

The number of events can go up to few 100s for each data protocol and we have around 4-5 data protocols. Metrics from different streams makes up a records 

like for example from above example for device 1 -

Device1 -> Metric1, Metric 2, Metric15 forms a single record for the device. Currently in development phase I am using interval join to achieve this, that is to create a record with latest data from different streams(events).

Based on data protocol I have 4-5 topics. Currently the data for a single event is being pushed to a partition of the kafka topic(producer key -> event_type + data_protocol). So essentially one topic is made up of many streams. I am filtering on the key to define the streams. 

My question is - Is this correct way to stream the data, I had thought of maintaining different topic for an event, however in that case number of topics could go to few thousands and that is something which becomes little challenging to maintain and not sure if kafka handles that well.

I know there are traditional ways to do this like pushing it to timeseries db and then joining data for different metric but that is something which will never scale, also this processing should be as realtime as possible.

Are there better ways to handle this use case or I am on correct path.

Thanks,

Hemant

[r_khachatryan]

Hello Hemant,

Thanks for your reply.

I think partitioning you described (event type + protocol type) is subject to data skew. Including a device ID should solve this problem.

Also, including "protocol_type" into the key and having topic per protocol_type seems redundant.

Furthermore, do you have any particular reason to maintain multiple topics? 

I could imagine protocols have different speeds or other characteristics, so you can tune Flink accordingly.

Otherwise, having a single topic partitioned only by device ID would simplify deployment and reduce data skew.

> By consume do you mean the downstream system?

Yes.

On Mon, May 11, 2020 at 11:30 PM hemant singh <[hidden email]> wrote:

Hello Roman,

PFB my response -

As I understand, each protocol has a distinct set of event types (where event type == metrics type); and a distinct set of devices. Is this correct?

Yes, correct. distinct events and devices. Each device emits these event.

> Based on data protocol I have 4-5 topics. Currently the data for a single event is being pushed to a partition of the kafka topic(producer key -> event_type + data_protocol).

Here you are talking about the source (to Flink job), right?

Yes, you are right.

Can you also share how are you going to consume these data?

By consume do you mean the downstream system? 

If yes then this data will be written to a DB, some metrics goes to TSDB(Influx) as well. 

Thanks,

Hemant

On Tue, May 12, 2020 at 2:08 AM Khachatryan Roman <[hidden email]> wrote:

Hi Hemant,

As I understand, each protocol has a distinct set of event types (where event type == metrics type); and a distinct set of devices. Is this correct?

> Based on data protocol I have 4-5 topics. Currently the data for a single event is being pushed to a partition of the kafka topic(producer key -> event_type + data_protocol).

Here you are talking about the source (to Flink job), right?

Can you also share how are you going to consume these data?

Regards,
Roman

On Mon, May 11, 2020 at 8:57 PM hemant singh <[hidden email]> wrote:

Hi,

I have different events from a device which constitutes different metrics for same device. Each of these event is produced by the device in interval of few milli seconds to a minute.

Event1(Device1) -> Stream1 -> Metric 1

Event2 (Device1) -> Stream2 -> Metric 2 ... 

..............

.......

Event100(Device1) -> Stream100 -> Metric100

The number of events can go up to few 100s for each data protocol and we have around 4-5 data protocols. Metrics from different streams makes up a records 

like for example from above example for device 1 -

Device1 -> Metric1, Metric 2, Metric15 forms a single record for the device. Currently in development phase I am using interval join to achieve this, that is to create a record with latest data from different streams(events).

Based on data protocol I have 4-5 topics. Currently the data for a single event is being pushed to a partition of the kafka topic(producer key -> event_type + data_protocol). So essentially one topic is made up of many streams. I am filtering on the key to define the streams. 

My question is - Is this correct way to stream the data, I had thought of maintaining different topic for an event, however in that case number of topics could go to few thousands and that is something which becomes little challenging to maintain and not sure if kafka handles that well.

I know there are traditional ways to do this like pushing it to timeseries db and then joining data for different metric but that is something which will never scale, also this processing should be as realtime as possible.

Are there better ways to handle this use case or I am on correct path.

Thanks,

Hemant

[hemant singh]

Hello Roman,

Thanks for your response.

I think partitioning you described (event type + protocol type) is subject to data skew. Including a device ID should solve this problem.

Also, including "protocol_type" into the key and having topic per protocol_type seems redundant.

Each protocol is in single topic and event_type is key to distribute data to a specific partition.

Furthermore, do you have any particular reason to maintain multiple topics? 

I could imagine protocols have different speeds or other characteristics, so you can tune Flink accordingly.

Otherwise, having a single topic partitioned only by device ID would simplify deployment and reduce data skew.

Yes, you are right. These protocols have separate characteristics like speed, data format. If I do have only one topic with data partitioned by device_id then it could be that events from faster protocol is processed faster and the joins which I want to do will not have enough matching data.

I have a question here how are you referring to tune Flink to handle different characteristics like speed of streams as reading from kafka could result in uneven processing of data?

> By consume do you mean the downstream system?

My downstream is TSDB and other DBs where the data will be written to. All these is time-series data.

Thanks,

Hemant

On Tue, May 12, 2020 at 5:28 PM Khachatryan Roman <[hidden email]> wrote:

Hello Hemant,

Thanks for your reply.

I think partitioning you described (event type + protocol type) is subject to data skew. Including a device ID should solve this problem.

Also, including "protocol_type" into the key and having topic per protocol_type seems redundant.

Furthermore, do you have any particular reason to maintain multiple topics? 

I could imagine protocols have different speeds or other characteristics, so you can tune Flink accordingly.

Otherwise, having a single topic partitioned only by device ID would simplify deployment and reduce data skew.

> By consume do you mean the downstream system?

Yes.

Regards,
Roman

On Mon, May 11, 2020 at 11:30 PM hemant singh <[hidden email]> wrote:

Hello Roman,

PFB my response -

As I understand, each protocol has a distinct set of event types (where event type == metrics type); and a distinct set of devices. Is this correct?

Yes, correct. distinct events and devices. Each device emits these event.

> Based on data protocol I have 4-5 topics. Currently the data for a single event is being pushed to a partition of the kafka topic(producer key -> event_type + data_protocol).

Here you are talking about the source (to Flink job), right?

Yes, you are right.

Can you also share how are you going to consume these data?

By consume do you mean the downstream system? 

If yes then this data will be written to a DB, some metrics goes to TSDB(Influx) as well. 

Thanks,

Hemant

On Tue, May 12, 2020 at 2:08 AM Khachatryan Roman <[hidden email]> wrote:

Hi Hemant,

As I understand, each protocol has a distinct set of event types (where event type == metrics type); and a distinct set of devices. Is this correct?

> Based on data protocol I have 4-5 topics. Currently the data for a single event is being pushed to a partition of the kafka topic(producer key -> event_type + data_protocol).

Here you are talking about the source (to Flink job), right?

Can you also share how are you going to consume these data?

Regards,
Roman

On Mon, May 11, 2020 at 8:57 PM hemant singh <[hidden email]> wrote:

Hi,

I have different events from a device which constitutes different metrics for same device. Each of these event is produced by the device in interval of few milli seconds to a minute.

Event1(Device1) -> Stream1 -> Metric 1

Event2 (Device1) -> Stream2 -> Metric 2 ... 

..............

.......

Event100(Device1) -> Stream100 -> Metric100

The number of events can go up to few 100s for each data protocol and we have around 4-5 data protocols. Metrics from different streams makes up a records 

like for example from above example for device 1 -

Device1 -> Metric1, Metric 2, Metric15 forms a single record for the device. Currently in development phase I am using interval join to achieve this, that is to create a record with latest data from different streams(events).

Based on data protocol I have 4-5 topics. Currently the data for a single event is being pushed to a partition of the kafka topic(producer key -> event_type + data_protocol). So essentially one topic is made up of many streams. I am filtering on the key to define the streams. 

My question is - Is this correct way to stream the data, I had thought of maintaining different topic for an event, however in that case number of topics could go to few thousands and that is something which becomes little challenging to maintain and not sure if kafka handles that well.

I know there are traditional ways to do this like pushing it to timeseries db and then joining data for different metric but that is something which will never scale, also this processing should be as realtime as possible.

Are there better ways to handle this use case or I am on correct path.

Thanks,

Hemant

[hemant singh]

Hello Flink Users,

Any views on this question of mine.

Thanks,

Hemant

On Tue, May 12, 2020 at 7:00 PM hemant singh <[hidden email]> wrote:

Hello Roman,

Thanks for your response.

I think partitioning you described (event type + protocol type) is subject to data skew. Including a device ID should solve this problem.

Also, including "protocol_type" into the key and having topic per protocol_type seems redundant.

Each protocol is in single topic and event_type is key to distribute data to a specific partition.

Furthermore, do you have any particular reason to maintain multiple topics? 

I could imagine protocols have different speeds or other characteristics, so you can tune Flink accordingly.

Otherwise, having a single topic partitioned only by device ID would simplify deployment and reduce data skew.

Yes, you are right. These protocols have separate characteristics like speed, data format. If I do have only one topic with data partitioned by device_id then it could be that events from faster protocol is processed faster and the joins which I want to do will not have enough matching data.

I have a question here how are you referring to tune Flink to handle different characteristics like speed of streams as reading from kafka could result in uneven processing of data?

> By consume do you mean the downstream system?

My downstream is TSDB and other DBs where the data will be written to. All these is time-series data.

Thanks,

Hemant

On Tue, May 12, 2020 at 5:28 PM Khachatryan Roman <[hidden email]> wrote:

Hello Hemant,

Thanks for your reply.

I think partitioning you described (event type + protocol type) is subject to data skew. Including a device ID should solve this problem.

Also, including "protocol_type" into the key and having topic per protocol_type seems redundant.

Furthermore, do you have any particular reason to maintain multiple topics? 

I could imagine protocols have different speeds or other characteristics, so you can tune Flink accordingly.

Otherwise, having a single topic partitioned only by device ID would simplify deployment and reduce data skew.

> By consume do you mean the downstream system?

Yes.

Regards,
Roman

On Mon, May 11, 2020 at 11:30 PM hemant singh <[hidden email]> wrote:

Hello Roman,

PFB my response -

As I understand, each protocol has a distinct set of event types (where event type == metrics type); and a distinct set of devices. Is this correct?

Yes, correct. distinct events and devices. Each device emits these event.

> Based on data protocol I have 4-5 topics. Currently the data for a single event is being pushed to a partition of the kafka topic(producer key -> event_type + data_protocol).

Here you are talking about the source (to Flink job), right?

Yes, you are right.

Can you also share how are you going to consume these data?

By consume do you mean the downstream system? 

If yes then this data will be written to a DB, some metrics goes to TSDB(Influx) as well. 

Thanks,

Hemant

On Tue, May 12, 2020 at 2:08 AM Khachatryan Roman <[hidden email]> wrote:

Hi Hemant,

As I understand, each protocol has a distinct set of event types (where event type == metrics type); and a distinct set of devices. Is this correct?

> Based on data protocol I have 4-5 topics. Currently the data for a single event is being pushed to a partition of the kafka topic(producer key -> event_type + data_protocol).

Here you are talking about the source (to Flink job), right?

Can you also share how are you going to consume these data?

Regards,
Roman

On Mon, May 11, 2020 at 8:57 PM hemant singh <[hidden email]> wrote:

Hi,

I have different events from a device which constitutes different metrics for same device. Each of these event is produced by the device in interval of few milli seconds to a minute.

Event1(Device1) -> Stream1 -> Metric 1

Event2 (Device1) -> Stream2 -> Metric 2 ... 

..............

.......

Event100(Device1) -> Stream100 -> Metric100

The number of events can go up to few 100s for each data protocol and we have around 4-5 data protocols. Metrics from different streams makes up a records 

like for example from above example for device 1 -

Device1 -> Metric1, Metric 2, Metric15 forms a single record for the device. Currently in development phase I am using interval join to achieve this, that is to create a record with latest data from different streams(events).

Based on data protocol I have 4-5 topics. Currently the data for a single event is being pushed to a partition of the kafka topic(producer key -> event_type + data_protocol). So essentially one topic is made up of many streams. I am filtering on the key to define the streams. 

My question is - Is this correct way to stream the data, I had thought of maintaining different topic for an event, however in that case number of topics could go to few thousands and that is something which becomes little challenging to maintain and not sure if kafka handles that well.

I know there are traditional ways to do this like pushing it to timeseries db and then joining data for different metric but that is something which will never scale, also this processing should be as realtime as possible.

Are there better ways to handle this use case or I am on correct path.

Thanks,

Hemant

[Arvid Heise-3]

Hi Hemant,

In general, you want to keep all data coming from one device in one Kafka partition, such that the timestamps of that device are monotonically increasing. Thus, when processing data from one device, you have ensured that no out-of-order events with respect to this device happen.

If you now want to aggregate all events of a given timestamp for a device, it is a matter of keying by device id and applying a custom window. There is no need for joins.

On Tue, May 12, 2020 at 9:05 PM hemant singh <[hidden email]> wrote:

Hello Flink Users,

Any views on this question of mine.

Thanks,

Hemant

On Tue, May 12, 2020 at 7:00 PM hemant singh <[hidden email]> wrote:

Hello Roman,

Thanks for your response.

I think partitioning you described (event type + protocol type) is subject to data skew. Including a device ID should solve this problem.

Also, including "protocol_type" into the key and having topic per protocol_type seems redundant.

Each protocol is in single topic and event_type is key to distribute data to a specific partition.

Furthermore, do you have any particular reason to maintain multiple topics? 

I could imagine protocols have different speeds or other characteristics, so you can tune Flink accordingly.

Otherwise, having a single topic partitioned only by device ID would simplify deployment and reduce data skew.

Yes, you are right. These protocols have separate characteristics like speed, data format. If I do have only one topic with data partitioned by device_id then it could be that events from faster protocol is processed faster and the joins which I want to do will not have enough matching data.

I have a question here how are you referring to tune Flink to handle different characteristics like speed of streams as reading from kafka could result in uneven processing of data?

> By consume do you mean the downstream system?

My downstream is TSDB and other DBs where the data will be written to. All these is time-series data.

Thanks,

Hemant

On Tue, May 12, 2020 at 5:28 PM Khachatryan Roman <[hidden email]> wrote:

Hello Hemant,

Thanks for your reply.

I think partitioning you described (event type + protocol type) is subject to data skew. Including a device ID should solve this problem.

Also, including "protocol_type" into the key and having topic per protocol_type seems redundant.

Furthermore, do you have any particular reason to maintain multiple topics? 

I could imagine protocols have different speeds or other characteristics, so you can tune Flink accordingly.

Otherwise, having a single topic partitioned only by device ID would simplify deployment and reduce data skew.

> By consume do you mean the downstream system?

Yes.

Regards,
Roman

On Mon, May 11, 2020 at 11:30 PM hemant singh <[hidden email]> wrote:

Hello Roman,

PFB my response -

As I understand, each protocol has a distinct set of event types (where event type == metrics type); and a distinct set of devices. Is this correct?

Yes, correct. distinct events and devices. Each device emits these event.

> Based on data protocol I have 4-5 topics. Currently the data for a single event is being pushed to a partition of the kafka topic(producer key -> event_type + data_protocol).

Here you are talking about the source (to Flink job), right?

Yes, you are right.

Can you also share how are you going to consume these data?

By consume do you mean the downstream system? 

If yes then this data will be written to a DB, some metrics goes to TSDB(Influx) as well. 

Thanks,

Hemant

On Tue, May 12, 2020 at 2:08 AM Khachatryan Roman <[hidden email]> wrote:

Hi Hemant,

As I understand, each protocol has a distinct set of event types (where event type == metrics type); and a distinct set of devices. Is this correct?

> Based on data protocol I have 4-5 topics. Currently the data for a single event is being pushed to a partition of the kafka topic(producer key -> event_type + data_protocol).

Here you are talking about the source (to Flink job), right?

Can you also share how are you going to consume these data?

Regards,
Roman

On Mon, May 11, 2020 at 8:57 PM hemant singh <[hidden email]> wrote:

Hi,

I have different events from a device which constitutes different metrics for same device. Each of these event is produced by the device in interval of few milli seconds to a minute.

Event1(Device1) -> Stream1 -> Metric 1

Event2 (Device1) -> Stream2 -> Metric 2 ... 

..............

.......

Event100(Device1) -> Stream100 -> Metric100

The number of events can go up to few 100s for each data protocol and we have around 4-5 data protocols. Metrics from different streams makes up a records 

like for example from above example for device 1 -

Device1 -> Metric1, Metric 2, Metric15 forms a single record for the device. Currently in development phase I am using interval join to achieve this, that is to create a record with latest data from different streams(events).

Based on data protocol I have 4-5 topics. Currently the data for a single event is being pushed to a partition of the kafka topic(producer key -> event_type + data_protocol). So essentially one topic is made up of many streams. I am filtering on the key to define the streams. 

My question is - Is this correct way to stream the data, I had thought of maintaining different topic for an event, however in that case number of topics could go to few thousands and that is something which becomes little challenging to maintain and not sure if kafka handles that well.

I know there are traditional ways to do this like pushing it to timeseries db and then joining data for different metric but that is something which will never scale, also this processing should be as realtime as possible.

Are there better ways to handle this use case or I am on correct path.

Thanks,

Hemant

--

[hemant singh]

Hi Arvid,

I don't want to aggregate all events, rather I want to create a record for a device combining data from multiple events. Each of this event gives me a metric for a device, so for example if I want one record for device-id=1 the metric will look like metric1, metric2, metric3....  where metric1 comes from a event1, metric2 from event2 and likewise....

From each event get latest data to form a kind of snapshot of device performance across the metrics.

Thanks,

Hemant

On Wed, May 13, 2020 at 1:38 AM Arvid Heise <[hidden email]> wrote:

Hi Hemant,

In general, you want to keep all data coming from one device in one Kafka partition, such that the timestamps of that device are monotonically increasing. Thus, when processing data from one device, you have ensured that no out-of-order events with respect to this device happen.

If you now want to aggregate all events of a given timestamp for a device, it is a matter of keying by device id and applying a custom window. There is no need for joins.

On Tue, May 12, 2020 at 9:05 PM hemant singh <[hidden email]> wrote:

Hello Flink Users,

Any views on this question of mine.

Thanks,

Hemant

On Tue, May 12, 2020 at 7:00 PM hemant singh <[hidden email]> wrote:

Hello Roman,

Thanks for your response.

I think partitioning you described (event type + protocol type) is subject to data skew. Including a device ID should solve this problem.

Also, including "protocol_type" into the key and having topic per protocol_type seems redundant.

Each protocol is in single topic and event_type is key to distribute data to a specific partition.

Furthermore, do you have any particular reason to maintain multiple topics? 

I could imagine protocols have different speeds or other characteristics, so you can tune Flink accordingly.

Otherwise, having a single topic partitioned only by device ID would simplify deployment and reduce data skew.

Yes, you are right. These protocols have separate characteristics like speed, data format. If I do have only one topic with data partitioned by device_id then it could be that events from faster protocol is processed faster and the joins which I want to do will not have enough matching data.

I have a question here how are you referring to tune Flink to handle different characteristics like speed of streams as reading from kafka could result in uneven processing of data?

> By consume do you mean the downstream system?

My downstream is TSDB and other DBs where the data will be written to. All these is time-series data.

Thanks,

Hemant

On Tue, May 12, 2020 at 5:28 PM Khachatryan Roman <[hidden email]> wrote:

Hello Hemant,

Thanks for your reply.

I think partitioning you described (event type + protocol type) is subject to data skew. Including a device ID should solve this problem.

Also, including "protocol_type" into the key and having topic per protocol_type seems redundant.

Furthermore, do you have any particular reason to maintain multiple topics? 

I could imagine protocols have different speeds or other characteristics, so you can tune Flink accordingly.

Otherwise, having a single topic partitioned only by device ID would simplify deployment and reduce data skew.

> By consume do you mean the downstream system?

Yes.

Regards,
Roman

On Mon, May 11, 2020 at 11:30 PM hemant singh <[hidden email]> wrote:

Hello Roman,

PFB my response -

As I understand, each protocol has a distinct set of event types (where event type == metrics type); and a distinct set of devices. Is this correct?

Yes, correct. distinct events and devices. Each device emits these event.

> Based on data protocol I have 4-5 topics. Currently the data for a single event is being pushed to a partition of the kafka topic(producer key -> event_type + data_protocol).

Here you are talking about the source (to Flink job), right?

Yes, you are right.

Can you also share how are you going to consume these data?

By consume do you mean the downstream system? 

If yes then this data will be written to a DB, some metrics goes to TSDB(Influx) as well. 

Thanks,

Hemant

On Tue, May 12, 2020 at 2:08 AM Khachatryan Roman <[hidden email]> wrote:

Hi Hemant,

As I understand, each protocol has a distinct set of event types (where event type == metrics type); and a distinct set of devices. Is this correct?

> Based on data protocol I have 4-5 topics. Currently the data for a single event is being pushed to a partition of the kafka topic(producer key -> event_type + data_protocol).

Here you are talking about the source (to Flink job), right?

Can you also share how are you going to consume these data?

Regards,
Roman

On Mon, May 11, 2020 at 8:57 PM hemant singh <[hidden email]> wrote:

Hi,

I have different events from a device which constitutes different metrics for same device. Each of these event is produced by the device in interval of few milli seconds to a minute.

Event1(Device1) -> Stream1 -> Metric 1

Event2 (Device1) -> Stream2 -> Metric 2 ... 

..............

.......

Event100(Device1) -> Stream100 -> Metric100

The number of events can go up to few 100s for each data protocol and we have around 4-5 data protocols. Metrics from different streams makes up a records 

like for example from above example for device 1 -

Device1 -> Metric1, Metric 2, Metric15 forms a single record for the device. Currently in development phase I am using interval join to achieve this, that is to create a record with latest data from different streams(events).

Based on data protocol I have 4-5 topics. Currently the data for a single event is being pushed to a partition of the kafka topic(producer key -> event_type + data_protocol). So essentially one topic is made up of many streams. I am filtering on the key to define the streams. 

My question is - Is this correct way to stream the data, I had thought of maintaining different topic for an event, however in that case number of topics could go to few thousands and that is something which becomes little challenging to maintain and not sure if kafka handles that well.

I know there are traditional ways to do this like pushing it to timeseries db and then joining data for different metric but that is something which will never scale, also this processing should be as realtime as possible.

Are there better ways to handle this use case or I am on correct path.

Thanks,

Hemant

--

Arvid Heise | Senior Java Developer

Follow us @VervericaData

--

Join Flink Forward - The Apache Flink Conference

Stream Processing | Event Driven | Real Time

--

Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany

--

Ververica GmbH
Registered at Amtsgericht Charlottenburg: HRB 158244 B
Managing Directors: Timothy Alexander Steinert, Yip Park Tung Jason, Ji (Toni) Cheng   

[Arvid Heise-3]

Hi Hemant,

what you described is an aggregation. You aggregate 15 small records into one large record. The concept of aggregation goes beyond pure numeric operations; for example, forming one large string with CONCAT is also a type of aggregation.

In your case, I'd still follow my general outlined approach. You have two options:

* Use an infinite window that fires at each new element. You need to ensure that old metrics are evicted [1].

* You can also implement a KeyedProcessFunction [2] and manage the state directly. It might be initially a bit more complex (=more code), but it should easier to understand and maintain.

[1] https://ci.apache.org/projects/flink/flink-docs-release-1.10/dev/stream/operators/windows.html#evictors

[2] https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/operators/process_function.html

On Wed, May 13, 2020 at 12:06 AM hemant singh <[hidden email]> wrote:

Hi Arvid,

I don't want to aggregate all events, rather I want to create a record for a device combining data from multiple events. Each of this event gives me a metric for a device, so for example if I want one record for device-id=1 the metric will look like metric1, metric2, metric3....  where metric1 comes from a event1, metric2 from event2 and likewise....

From each event get latest data to form a kind of snapshot of device performance across the metrics.

Thanks,

Hemant

On Wed, May 13, 2020 at 1:38 AM Arvid Heise <[hidden email]> wrote:

Hi Hemant,

In general, you want to keep all data coming from one device in one Kafka partition, such that the timestamps of that device are monotonically increasing. Thus, when processing data from one device, you have ensured that no out-of-order events with respect to this device happen.

If you now want to aggregate all events of a given timestamp for a device, it is a matter of keying by device id and applying a custom window. There is no need for joins.

On Tue, May 12, 2020 at 9:05 PM hemant singh <[hidden email]> wrote:

Hello Flink Users,

Any views on this question of mine.

Thanks,

Hemant

On Tue, May 12, 2020 at 7:00 PM hemant singh <[hidden email]> wrote:

Hello Roman,

Thanks for your response.

I think partitioning you described (event type + protocol type) is subject to data skew. Including a device ID should solve this problem.

Also, including "protocol_type" into the key and having topic per protocol_type seems redundant.

Each protocol is in single topic and event_type is key to distribute data to a specific partition.

Furthermore, do you have any particular reason to maintain multiple topics? 

I could imagine protocols have different speeds or other characteristics, so you can tune Flink accordingly.

Otherwise, having a single topic partitioned only by device ID would simplify deployment and reduce data skew.

Yes, you are right. These protocols have separate characteristics like speed, data format. If I do have only one topic with data partitioned by device_id then it could be that events from faster protocol is processed faster and the joins which I want to do will not have enough matching data.

I have a question here how are you referring to tune Flink to handle different characteristics like speed of streams as reading from kafka could result in uneven processing of data?

> By consume do you mean the downstream system?

My downstream is TSDB and other DBs where the data will be written to. All these is time-series data.

Thanks,

Hemant

On Tue, May 12, 2020 at 5:28 PM Khachatryan Roman <[hidden email]> wrote:

Hello Hemant,

Thanks for your reply.

I think partitioning you described (event type + protocol type) is subject to data skew. Including a device ID should solve this problem.

Also, including "protocol_type" into the key and having topic per protocol_type seems redundant.

Furthermore, do you have any particular reason to maintain multiple topics? 

I could imagine protocols have different speeds or other characteristics, so you can tune Flink accordingly.

Otherwise, having a single topic partitioned only by device ID would simplify deployment and reduce data skew.

> By consume do you mean the downstream system?

Yes.

Regards,
Roman

On Mon, May 11, 2020 at 11:30 PM hemant singh <[hidden email]> wrote:

Hello Roman,

PFB my response -

As I understand, each protocol has a distinct set of event types (where event type == metrics type); and a distinct set of devices. Is this correct?

Yes, correct. distinct events and devices. Each device emits these event.

> Based on data protocol I have 4-5 topics. Currently the data for a single event is being pushed to a partition of the kafka topic(producer key -> event_type + data_protocol).

Here you are talking about the source (to Flink job), right?

Yes, you are right.

Can you also share how are you going to consume these data?

By consume do you mean the downstream system? 

If yes then this data will be written to a DB, some metrics goes to TSDB(Influx) as well. 

Thanks,

Hemant

On Tue, May 12, 2020 at 2:08 AM Khachatryan Roman <[hidden email]> wrote:

Hi Hemant,

As I understand, each protocol has a distinct set of event types (where event type == metrics type); and a distinct set of devices. Is this correct?

> Based on data protocol I have 4-5 topics. Currently the data for a single event is being pushed to a partition of the kafka topic(producer key -> event_type + data_protocol).

Here you are talking about the source (to Flink job), right?

Can you also share how are you going to consume these data?

Regards,
Roman

On Mon, May 11, 2020 at 8:57 PM hemant singh <[hidden email]> wrote:

Hi,

I have different events from a device which constitutes different metrics for same device. Each of these event is produced by the device in interval of few milli seconds to a minute.

Event1(Device1) -> Stream1 -> Metric 1

Event2 (Device1) -> Stream2 -> Metric 2 ... 

..............

.......

Event100(Device1) -> Stream100 -> Metric100

The number of events can go up to few 100s for each data protocol and we have around 4-5 data protocols. Metrics from different streams makes up a records 

like for example from above example for device 1 -

Device1 -> Metric1, Metric 2, Metric15 forms a single record for the device. Currently in development phase I am using interval join to achieve this, that is to create a record with latest data from different streams(events).

Based on data protocol I have 4-5 topics. Currently the data for a single event is being pushed to a partition of the kafka topic(producer key -> event_type + data_protocol). So essentially one topic is made up of many streams. I am filtering on the key to define the streams. 

My question is - Is this correct way to stream the data, I had thought of maintaining different topic for an event, however in that case number of topics could go to few thousands and that is something which becomes little challenging to maintain and not sure if kafka handles that well.

I know there are traditional ways to do this like pushing it to timeseries db and then joining data for different metric but that is something which will never scale, also this processing should be as realtime as possible.

Are there better ways to handle this use case or I am on correct path.

Thanks,

Hemant

--

Arvid Heise | Senior Java Developer

Follow us @VervericaData

--

Join Flink Forward - The Apache Flink Conference

Stream Processing | Event Driven | Real Time

--

Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany

--

Ververica GmbH
Registered at Amtsgericht Charlottenburg: HRB 158244 B
Managing Directors: Timothy Alexander Steinert, Yip Park Tung Jason, Ji (Toni) Cheng   

--