title: Spark Streaming not performing operations on read blocks
body: <p>I am newbie to Spark Streaming concept and have been stuck from last two days trying to understand Spark streaming from socket. I see Spark is able to read blocks passed to the socket. However it does not perform any operation on the read blocks. </p>
  
  <p>Here is the Spark code</p>
  
  <pre><code>package foo;
  import java.io.File;
  import java.util.Arrays;
  import java.util.LinkedList;
  import java.util.List;
  
  import org.apache.spark.SparkConf;
  import org.apache.spark.api.java.JavaRDD;
  import org.apache.spark.api.java.JavaSparkContext;
  import org.apache.spark.api.java.function.Function;
  import org.apache.spark.api.java.function.Function2;
  import org.apache.spark.api.java.function.PairFunction;
  import org.apache.spark.streaming.api.java.JavaDStream;
  import org.apache.spark.streaming.api.java.JavaPairDStream;
  import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
  import org.apache.spark.streaming.api.java.JavaStreamingContext;
  
  import scala.Tuple2;
  
  public class AppSocket {
  
  public static void main(String[] args)
  {
  
      SparkConf conf = new SparkConf().setAppName("KAFKA").setMaster("local");
  
      JavaStreamingContext jssc = new JavaStreamingContext(conf, new org.apache.spark.streaming.Duration(1000));
  
      JavaReceiverInputDStream&lt;String&gt; inputStream = jssc.socketTextStream("localhost", 33333);
  
  
      JavaPairDStream&lt;String, Integer&gt; mappedStream = inputStream.mapToPair(
          new PairFunction&lt;String, String, Integer&gt;() {
  
            public Tuple2&lt;String, Integer&gt; call(String i) {
               System.out.println(i);
              return new Tuple2&lt;String, Integer&gt;(i , 1);
            }
          });
  
      JavaPairDStream&lt;String, Integer&gt; reducedStream = mappedStream.reduceByKey(
        new Function2&lt;Integer, Integer, Integer&gt;() {
  
          public Integer call(Integer i1, Integer i2) {
            return i1 + i2;
          }
      });
  
      reducedStream.print();
      System.out.println("Testing........"+reducedStream.count());
      jssc.start();
      jssc.awaitTermination();
  
  
  }
  
  }
  </code></pre>
  
  <p>I am running netcat to create output stream on specified port</p>
  
  <pre><code>nc -lk 33333
  </code></pre>
  
  <p>I have tried to create output stream. Here is my java code</p>
  
  <pre><code>    ServerSocket serverSocket = null;
      int portNumber = 33333;
      serverSocket = new ServerSocket(portNumber);
  
      System.out.println("Server Waiting.................");
  
      Socket clientSocket = serverSocket.accept();
  
      System.out.println("Server Connected!!!!!!!!!!!!!!!");
  
  
      // Wait for a message
      int countflag = 0;
      PrintWriter out = null;
      out = new PrintWriter(clientSocket.getOutputStream(), true);
      while(true)
      {
      Message message = consumer.receive(1000);
  
      if (message instanceof TextMessage) {
          TextMessage textMessage = (TextMessage) message;
          String text = textMessage.getText();
          System.out.println("Received: " + text);
          list.add(text);
          System.out.println(++countflag);
          if(list.size() &gt; 50)
          {
  
          for(int i = 0; i &lt; list.size() ; i++)
          {
         System.out.print(i);
          out.write(text);
          out.write("\n");
          out.flush();
          }
  
         list.clear();
          }
  
      } else {
          count++;
  
      }
      if(count &gt; 100) break;
      }
      out.close();
      consumer.close();
      session.close();
      connection.close();
  </code></pre>
  
  <p>Spark consumes the block sent on the stream however it does not perform any action required on the streamed blocks.</p>
  
  <p>Spark Output Console</p>
  
  <pre><code>14/11/26 15:32:14 INFO MemoryStore: ensureFreeSpace(12) called with curMem=3521, maxMem=278302556
  14/11/26 15:32:14 INFO MemoryStore: Block input-0-1417015934400 stored as bytes in memory (estimated size 12.0 B, free 265.4 MB)
  14/11/26 15:32:14 INFO BlockManagerInfo: Added input-0-1417015934400 in memory on ip-10-0-1-56.ec2.internal:57275 (size: 12.0 B, free: 265.4 MB)
  14/11/26 15:32:14 INFO BlockManagerMaster: Updated info of block input-0-1417015934400
  14/11/26 15:32:14 WARN BlockManager: Block input-0-1417015934400 already exists on this machine; not re-adding it
  14/11/26 15:32:14 INFO BlockGenerator: Pushed block input-0-1417015934400
  14/11/26 15:32:15 INFO ReceiverTracker: Stream 0 received 1 blocks
  14/11/26 15:32:15 INFO JobScheduler: Added jobs for time 1417015935000 ms
  </code></pre>
  
  <p>I would appreciate your help. Thanks in advance</p>