You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by Ilya Karpov <id...@gmail.com> on 2024/06/17 11:39:13 UTC
A way to meter number of deserialization errors
Hi all,
we are planning to use flink as a connector between kafka and
external systems. We use protobuf as a message format in kafka. If
non-backward compatible changes occur we want to skip those messages
('protobuf.ignore-parse-errors' = 'true') but record an error and raise an
alert. I didn't find any way to record deserialization errors (
https://github.com/apache/flink/blob/3a15d1ce69ac21d619f60033ec45cae303489c8f/flink-formats/flink-protobuf/src/main/java/org/apache/flink/formats/protobuf/deserialize/PbRowDataDeserializationSchema.java#L73),
does it exist?
Re: A way to meter number of deserialization errors
Posted by Ilya Karpov <id...@gmail.com>.
Does anybody experience the problem of metering deserialization errors?
пн, 17 июн. 2024 г. в 14:39, Ilya Karpov <id...@gmail.com>:
> Hi all,
> we are planning to use flink as a connector between kafka and
> external systems. We use protobuf as a message format in kafka. If
> non-backward compatible changes occur we want to skip those messages
> ('protobuf.ignore-parse-errors' = 'true') but record an error and raise an
> alert. I didn't find any way to record deserialization errors (
> https://github.com/apache/flink/blob/3a15d1ce69ac21d619f60033ec45cae303489c8f/flink-formats/flink-protobuf/src/main/java/org/apache/flink/formats/protobuf/deserialize/PbRowDataDeserializationSchema.java#L73),
> does it exist?
>
Re: A way to meter number of deserialization errors
Posted by Ilya Karpov <id...@gmail.com>.
I guess metering deserialization errors may be done by flink metrics,
but now it is missing out of the box.
> - I wondered if you might also want to count how many were successfully parsed in a non-protobuf layer (Dynamic table sort of level)
No, this is not a requirement. I have a simple flink-sql job that
streams proto-messages from kafka to hive, with quite simple mapping.
> - do you know why some messages do not parse? It would interesting to know the types of non-parsable messages, I assume with the idea to understand and potentially fix the source / parser limitations.
Because of a human mistake: developers can introduce non backward
compatible changes in the producer, emitted proto-messages will not be
deserialized on a flink-app side. Failing flink-job actually is an
option worth considering (because now it's the only one source of
proto-messages in the topic and it's ok to halt-and-repair).
ср, 19 июн. 2024 г. в 11:39, David Radley <da...@uk.ibm.com>:
>
> Hi Ilya,
>
> I have not got any experience of doing this, but wonder if we could use the Flink Metrics . I wonder:
>
> - There could be hook point at that part of the code to discover some custom code that implements the metrics.
>
> - A better way might be to throw a new unable-to-parse Exception that could be caught in non-protobuf code and then a count metric be incremented. I guess the metrics would be scoped to a Job – like the watermark metrics.
>
> - I wondered if you might also want to count how many were successfully parsed in a non-protobuf layer (Dynamic table sort of level)
>
> - do you know why some messages do not parse? It would interesting to know the types of non-parsable messages, I assume with the idea to understand and potentially fix the source / parser limitations.
>
>
>
> Kind regards, David.
>
>
>
> From: Ilya Karpov <id...@gmail.com>
> Date: Monday, 17 June 2024 at 12:39
> To: user <us...@flink.apache.org>
> Subject: [EXTERNAL] A way to meter number of deserialization errors
>
> Hi all, we are planning to use flink as a connector between kafka and external systems. We use protobuf as a message format in kafka. If non-backward compatible changes occur we want to skip those messages ('protobuf. ignore-parse-errors'
>
> Hi all,
>
> we are planning to use flink as a connector between kafka and external systems. We use protobuf as a message format in kafka. If non-backward compatible changes occur we want to skip those messages ('protobuf.ignore-parse-errors' = 'true') but record an error and raise an alert. I didn't find any way to record deserialization errors (https://github.com/apache/flink/blob/3a15d1ce69ac21d619f60033ec45cae303489c8f/flink-formats/flink-protobuf/src/main/java/org/apache/flink/formats/protobuf/deserialize/PbRowDataDeserializationSchema.java#L73), does it exist?
>
> Unless otherwise stated above:
>
> IBM United Kingdom Limited
> Registered in England and Wales with number 741598
> Registered office: PO Box 41, North Harbour, Portsmouth, Hants. PO6 3AU
Re: A way to meter number of deserialization errors
Posted by David Radley <da...@uk.ibm.com>.
Hi Ilya,
I have not got any experience of doing this, but wonder if we could use the Flink Metrics <https://nightlies.apache.org/flink/flink-docs-master/docs/ops/metrics/> . I wonder:
- There could be hook point at that part of the code to discover some custom code that implements the metrics.
- A better way might be to throw a new unable-to-parse Exception that could be caught in non-protobuf code and then a count metric be incremented. I guess the metrics would be scoped to a Job – like the watermark metrics.
- I wondered if you might also want to count how many were successfully parsed in a non-protobuf layer (Dynamic table sort of level)
- do you know why some messages do not parse? It would interesting to know the types of non-parsable messages, I assume with the idea to understand and potentially fix the source / parser limitations.
Kind regards, David.
From: Ilya Karpov <id...@gmail.com>
Date: Monday, 17 June 2024 at 12:39
To: user <us...@flink.apache.org>
Subject: [EXTERNAL] A way to meter number of deserialization errors
Hi all, we are planning to use flink as a connector between kafka and external systems. We use protobuf as a message format in kafka. If non-backward compatible changes occur we want to skip those messages ('protobuf. ignore-parse-errors'
Hi all,
we are planning to use flink as a connector between kafka and external systems. We use protobuf as a message format in kafka. If non-backward compatible changes occur we want to skip those messages ('protobuf.ignore-parse-errors' = 'true') but record an error and raise an alert. I didn't find any way to record deserialization errors (https://github.com/apache/flink/blob/3a15d1ce69ac21d619f60033ec45cae303489c8f/flink-formats/flink-protobuf/src/main/java/org/apache/flink/formats/protobuf/deserialize/PbRowDataDeserializationSchema.java#L73), does it exist?
Unless otherwise stated above:
IBM United Kingdom Limited
Registered in England and Wales with number 741598
Registered office: PO Box 41, North Harbour, Portsmouth, Hants. PO6 3AU