Hello I am trying to find the best way to rename Json Key fi Apache Flink #troubleshooting

Hello I am trying to find the best way to rename J...

Zachary Schmerber

04/24/2023, 5:44 PM

Hello I am trying to find the best way to rename Json Key fields and leave the Values the same. I have already turned the datastream into an ObjectNode. What is the best way to accomplish the renaming of the Key(log field name) at this point?

Copy code

// log source data from above Kafka Topic using JsonDeserializationSchema//

        DataStream<ObjectNode> filteredStream = env.fromSource(logSource, WatermarkStrategy.noWatermarks(), "Kafka winlog Source");

        DataStream<ObjectNode> winObjectStream = filteredStream.filter(node -> node.get("service_id").asText().equals("cdd31c3f-3b24-4526-848a-xxxxxxxxxxx"));


   //Placeholder for logically mapping the new Object Stream 

// Datastream that has sucessfully been transformed into Objects 
        winObjectStream.sinkTo(winEventSink);

David Anderson

04/25/2023, 3:41 AM

Well, for starters you probably don't want to use a

DataStream<ObjectNode>

. That's going to be painfully inefficient. I would suggest you use a JsonDeserializationSchema to deserialize the JSON into a POJO, if possible. But as for your question, I confess I don't understand what you're trying to accomplish.

Zachary Schmerber

04/25/2023, 4:04 AM

Basically I have an a Kafka topic with windowsEventlogs in JSON format. We have many other logs in the same Topic. I need to filter out the windows logs and rename many of the field names to match OCSF schema then reserialize the data into a Kafka Sink in JSON again. Given our data volume i am assuming DataStream<ObjectNode> is not going to work.

Zachary Schmerber

04/25/2023, 4:13 AM

In order to filter out the winlogs without using DataStream<ObjectNode>.. i tried SimpleStringSchema for the filter and applied the POJO with custom Deseralizationschema: https://github.com/zschemrber/secint-etl-flink-job/blob/main/src/main/java/org/atlassian/TransformationJob.java Now that i have a stream with the DataStream<Message> I do not know how to rename the key field names to meet new OCSF naming convention. Is that something I do during the re serialization phase of this ?

Zachary Schmerber

04/25/2023, 4:54 PM

I am guessing this is my best option? https://www.javaguides.net/2019/04/change-field-name-in-json-using-jackson.html

David Anderson

04/26/2023, 4:25 PM

You could introduce a second POJO type with the final field names, and use a map to transform the POJOs that match the input schema into the POJOs that correspond to the desired output schema.

Zachary Schmerber

04/26/2023, 4:29 PM

This could be a good way to go. Do you have an example of this i can use for reference?

Zachary Schmerber

04/27/2023, 9:40 PM

I am thinking about using this to generate the Source Pojo, then doing the same for OCSF schema server api for the second pojo generation. Once they are both generated i can make the mapping. https://github.com/joelittlejohn/jsonschema2pojo/blob/master/README.md

Open in Slack

Previous Next