Pfthb

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.<Long, String>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<KV<Long, String>>
 )
 .apply(Values.<String>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);

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.<String>create())
.apply(
 "JSONtoData", // the transform name
 ParDo.of(new DoFn<String, String>() // a DoFn as an anonymous inner class instance
 @ProcessElement
 public void processElement(@Element String word, OutputReceiver<String> 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<String, YourCSVClassHere>()
OutputReceiver<YourCSVClassHere> out

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