Thanks Kamil.
I'll give it a try.
Regards Hans
On Fri, Jan 14, 2022, 20:32 Kamil ty <kamilt...@gmail.com> wrote:
> Hello Hans,
>
> As far as I know the JSONKeyValueDeserializationSchema returns a Jackson
> ObjectNode. Below I have included an example based on Flink stable
> documentation.
>
> KafkaSource<ObjectNode> source = KafkaSource.<ObjectNode>builder()
> .setBootstrapServers(brokers)
> .setTopics("input-topic")
> .setGroupId("my-group")
> .setStartingOffsets(OffsetsInitializer.earliest())
> .setDeserializer(new JSONKeyValueDeserializationSchema(false))
> .build();
>
> DataStream<ObjectNode> ds = env.fromSource(source,
> WatermarkStrategy.noWatermarks(), "Kafka Source");
> // Below we access the JSON field stored in the ObjectNode.
> DataStream<String> processedDs = ds.map(record ->
> record.get("value").get("my-field").asText());
>
> It is also possible to implement your own deserialization schema that for
> eg. could turn each record into a POJO. You can do this by implementing the
> KafkaDeserializationSchema
> (Flink : 1.14-SNAPSHOT API) (apache.org)
> <https://nightlies.apache.org/flink/flink-docs-release-1.14/api/java/org/apache/flink/streaming/connectors/kafka/KafkaDeserializationSchema.html>.
> If you are only interested in the value of the Kafka record, you can also
> extend the AbstractDeserializationSchema (Flink : 1.14-SNAPSHOT API)
> (apache.org)
> <https://nightlies.apache.org/flink/flink-docs-stable/api/java/org/apache/flink/api/common/serialization/AbstractDeserializationSchema.html>
> and
> use .setValueOnlyDeserializer(new CustomDeserializer()). There is also a
> different API that you could use for this which is specified here:
> KafkaSourceBuilder
> (Flink : 1.14-SNAPSHOT API) (apache.org)
> <https://nightlies.apache.org/flink/flink-docs-stable/api/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.html>.
> Although the customDeserializer will be the same for older Flink versions,
> the Kafka Source has appeared recently, to learn about the previous kafka
> source (FlinkKafkaConsumer) see: Kafka | Apache Flink
> <https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/connectors/datastream/kafka/#kafka-sourcefunction>
> .
>
> Best Regards
> Kamil
>
> On Fri, 14 Jan 2022 at 18:37, HG <hanspeter.sl...@gmail.com> wrote:
>
>> Hi,
>>
>> Before starting programming myself I'd like to know whether there are
>> good examples with deserialization of JSON that I can borrow.
>> The structure of the JSON is nested with multiple levels.
>>
>> Any references?
>>
>> 'better well stolen than badly invented myself' we'd say in Dutch😁
>>
>> Regards Hans
>>
>
