title: Fastest way to sink JSON to Kafka using FLINK
body: <h1>Code Optimisation</h1>
  <p>I have a flink application that reads data from a url/port performs processing on it and returns a JSON. I then convert the JSON to a String and sink it to Kafka.</p>
  <h1>Current Performance &amp; Noted Issue</h1>
  <p>If I just perform the processing -&gt; I can run about 30,000 strings through the function, however when I add the function to convert it to STring and then sink to kafka My throughput drops to 17,000 strings per second.</p>
  <p>Do I need to convert my JSON to String before I sink to Kafka? If not how do I sink a json ObjectNode to kafka?</p>
  <p>Else what other solutions are there.
  I think the bottleneck is the to String Function</p>
  <p>I tried converting the JSON to a String using several methods (.toString function, StringBuilder to String).</p>
  <pre><code> // Read from Source
   val in_stream = env.socketTextStream(url, port,      socket_stream_deliminator, socket_connection_retries).setParallelism(1)
  
   // Perform Process
   .map(x=&gt;{Process(x)}).setParallelism(1)
  
   // Convert to STring
   .map(x =&gt; ObjectNodeToString({
       val json_string_builder = StringBuilder.newBuilder
       json_string_builder.append(x)
       return json_string_builder.toString()
   })).setParallelism(1)
  
   // sink data
   .addSink(new FlinkKafkaProducer[String](broker_hosts, global_topic, new SimpleStringSchema()))
  </code></pre>
  <p>I would like to mantain the 30,000 strings processing per second. which I do get with out the convert to string function. Can I sink the ObjectNode directly to kafka?</p>