title: How to get JSON Elements from String(full with JSON object).
body: <p>I have a JSON file</p>
  
  <pre><code>{ "impressions":
          [
            {
              "impressionId": "7ad7a77fas346a7a2a1da6",
              "userId": "hafsa",
              "clientId": "400"
            },
            {
              "impressionId": "7ad7a77fas346a7a2a1da7",
              "userId": "asif",
              "clientId": "200"
            },
            {
              "impressionId": "7ad7a77fas346a7a2a1da8",
              "userId": "zadarov",
              "clientId": "300"
            },
            {
              "impressionId": "7ad7a77fas346a7a2a1da9",
              "userId": "julia",
              "clientId": "100"
            }
          ]
  }
  </code></pre>
  
  <p>I am working on Kafka and have a NodeJS Producer and Java Consumer.I have to get each impression as a separate message in my consumer. My nodeJS code is:</p>
  
  <pre><code>console.log("Kafka is recieving JSON....");
  var fs = require('fs');
  var kafka = require('kafka-node');
  var Producer = kafka.Producer;
  var Client = kafka.Client;
  var client = new Client('localhost:2181');
  var producer = new Producer(client);
  
  
  fs.readFile('data.json', 'utf8', function (err, data) {
      if (err) throw err;
  
  
      var jsonobj = JSON.parse(data);
      var countMessages = jsonobj.impressions.length;
      var impressionArr = [];
      impressionArr = jsonobj.impressions;
      payloads = [
          { topic: 'test', messages: impressionArr, partition: 0 }
      ];
      producer.on('ready', function(){
          producer.send(payloads, function(err, data){
  
              console.log(data);
          });
      });
      producer.on('error', function(err){
          console.log("Error: "+err)
      });
  
  
  });
  </code></pre>
  
  <p>and my JAVA consumer is :</p>
  
  <pre><code>JavaInputDStream&lt;String&gt; messagesFrmSpecifiedOffset= KafkaUtils.createDirectStream(
                  sc,
                  String.class,
                  String.class,
                  StringDecoder.class,
                  StringDecoder.class,
                  String.class,
                  kafkaParams,
                  fromOffset,
                  new Function&lt;MessageAndMetadata&lt;String, String&gt;, String&gt;() {
                      public String call(MessageAndMetadata&lt;String, String&gt; msgAndMd) throws Exception {
  
                          return msgAndMd.message();
                      }
                  }
  
          );
  </code></pre>
  
  <p>and I am getting impressions as separate messages as follows in JAVA consumer.</p>
  
  <pre><code>JavaDStream&lt;Record&gt; rdd_impressionRecords = messagesFrmSpecifiedOffset.map(
                  new Function&lt;String, Record&gt;() {
                      public Record call(String impressions) throws Exception {
  
                          System.out.println("impressions: " + impressions);
  
                          Record sd = new Record();
                          return sd;
                      }
                  });
  </code></pre>
  
  <p>BUT MY PROBLEM IS:
  that I am getting output in object form like:</p>
  
  <pre><code>impressions: [object Object]
  impressions: [object Object]
  impressions: [object Object]
  impressions: [object Object]
  </code></pre>
  
  <p>I know that I am sending JSON and recieving as String but anyone can help me to do solution in three ways:
  1. Can I get key, values from [object Object] in Java class.
  2. Is it possible to send each array element as String (using JSON.stringify) in NodeJS without any loop. Direct transfer as I am doing.
  3. Can I rewrite method of createDirectStream that returns JSON obect directly.</p>