How to parse json in a beam pipeline?

Question

My data is of newline delimited json form and looks like shown below. I am reading this type of data from a Kafka topic.

{"sender":"S1","senderHost":"ip-10-20-30-40","timestamp":"2018-08-13T16:17:12.874Z","topic":"test","messageType":"type_1","data":{"name":"John Doe", "id":"12DROIY321"}}

I want to build an apache Beam pipeline which reads this data from Kafka, parses this json format to give me an output as shown below:

S1,2018-08-13T16:17:12.874Z,type_1,12DROIY321

The output is basically a comma delimited string consisting of the sender, timestamp, messageType and id from within data.

My code so far is as below:

public class Pipeline1{
    public static void main(String[] args){
        PipelineOptions options = PipelineOptionsFactory.create();

        // Create the Pipeline object with the options we defined above.
        Pipeline p = Pipeline.create(options);

        p.apply(KafkaIO.read()
                .withBootstrapServers("localhost:9092")
                .withTopic("test")
                .withKeyDeserializer(LongDeserializer.class)
                .withValueDeserializer(StringDeserializer.class)

                .updateConsumerProperties(ImmutableMap.of("auto.offset.reset", (Object)"earliest"))

                // We're writing to a file, which does not support unbounded data sources. This line makes it bounded to
                // the first 35 records.
                // In reality, we would likely be writing to a data source that supports unbounded data, such as BigQuery.
                .withMaxNumRecords(35)

                .withoutMetadata() // PCollection>
        )
                .apply(Values.create())
                .apply(TextIO.write().to("test"));

        p.run().waitUntilFinish();

    }
}

I am unable to figure out how to parse the json to get the required csv format within the pipeline. Using the code above, I am able to write the same json lines into a file, and using the code below, i can parse the json, but can anyone please help me figure out how to accomplish this as an additional step with the beam pipeline logic?

JSONParser parser = new JSONParser();
            Object obj = null;
            try {
                obj = parser.parse(strLine);
            } catch (ParseException e) {
                e.printStackTrace();
            }
            JSONObject jsonObject =  (JSONObject) obj;

            String sender = (String) jsonObject.get("sender");

            String messageType = (String) jsonObject.get("messageType");

            String timestamp = (String) jsonObject.get("timestamp");

            System.out.println(sender+","+timestamp+","+messageType);

user_5445883 · Answer

According to the documentation, you will need to write a transformation (or find one that matches your use case).

https://beam.apache.org/documentation/programming-guide/#composite-transforms

The documentation also provides an excellent example.

Example that should produce your output:

.apply(Values.create())
.apply(
    "JSONtoData",                     // the transform name
    ParDo.of(new DoFn() {    // a DoFn as an anonymous inner class instance
        @ProcessElement
        public void processElement(@Element String word, OutputReceiver out) {
            JSONParser parser = new JSONParser();
            Object obj = null;
            try {
                obj = parser.parse(strLine);
            } catch (ParseException e) {
                e.printStackTrace();
            }
            JSONObject jsonObject =  (JSONObject) obj;

            String sender = (String) jsonObject.get("sender");
            String messageType = (String) jsonObject.get("messageType");
            String timestamp = (String) jsonObject.get("timestamp");

            out.output(sender+","+timestamp+","+messageType);
        }
   }));

To return CSV values, just change the generics to:

new DoFn()
OutputReceiver out

I didn't test this code, use at own risk.

How to parse json in a beam pipeline?

1 Answers