JavaDStream print RDDs in lambda to console

Question

I am new to spark and i'm trying to create simple JavaDStream for testing my work with spark-testing-base API. what i have done so far was:

    JavaStreamingContext streamingContext = new 
          JavaStreamingContext(jsc(),Durations.seconds(10));
    List<String> list = new LinkedList<String>();
    list.add("first");
    list.add("second");
    list.add("third");
    JavaRDD<String> myVeryOwnRDD = jsc().parallelize(list);
    Queue<JavaRDD<String>> queue = new LinkedList<JavaRDD<String>>();
    queue.add( myVeryOwnRDD );
    JavaDStream<String> javaDStream = streamingContext.queueStream( queue );

    javaDStream.foreachRDD( x-> {
        x.collect().stream().forEach(n-> System.out.println("item of list: "+n));
    });

I was hopping that it will print my list.. it didn't. Insted of it i got:

12:19:05.454 [main] DEBUG org.apache.spark.util.ClosureCleaner - +++ Cleaning closure <function1> (org.apache.spark.streaming.api.java.JavaDStreamLike$$anonfun$foreachRDD$3) +++
12:19:05.468 [main] DEBUG org.apache.spark.util.ClosureCleaner -  + declared fields: 3
12:19:05.469 [main] DEBUG org.apache.spark.util.ClosureCleaner -      public static final long org.apache.spark.streaming.api.java.JavaDStreamLike$$anonfun$foreachRDD$3.serialVersionUID
12:19:05.469 [main] DEBUG org.apache.spark.util.ClosureCleaner -      private final org.apache.spark.streaming.api.java.JavaDStreamLike org.apache.spark.streaming.api.java.JavaDStreamLike$$anonfun$foreachRDD$3.$outer
12:19:05.469 [main] DEBUG org.apache.spark.util.ClosureCleaner -      private final org.apache.spark.api.java.function.VoidFunction org.apache.spark.streaming.api.java.JavaDStreamLike$$anonfun$foreachRDD$3.foreachFunc$3
12:19:05.469 [main] DEBUG org.apache.spark.util.ClosureCleaner -  + declared methods: 2
12:19:05.470 [main] DEBUG org.apache.spark.util.ClosureCleaner -      public final java.lang.Object org.apache.spark.streaming.api.java.JavaDStreamLike$$anonfun$foreachRDD$3.apply(java.lang.Object)
12:19:05.470 [main] DEBUG org.apache.spark.util.ClosureCleaner -      public final void org.apache.spark.streaming.api.java.JavaDStreamLike$$anonfun$foreachRDD$3.apply(org.apache.spark.rdd.RDD)
12:19:05.470 [main] DEBUG org.apache.spark.util.ClosureCleaner -  + inner classes: 0
12:19:05.471 [main] DEBUG org.apache.spark.util.ClosureCleaner -  + outer classes: 1
12:19:05.472 [main] DEBUG org.apache.spark.util.ClosureCleaner -      org.apache.spark.streaming.api.java.JavaDStreamLike
12:19:05.472 [main] DEBUG org.apache.spark.util.ClosureCleaner -  + outer objects: 1
12:19:05.473 [main] DEBUG org.apache.spark.util.ClosureCleaner -      org.apache.spark.streaming.api.java.JavaDStream@7209ffb5
12:19:05.474 [main] DEBUG org.apache.spark.util.ClosureCleaner -  + populating accessed fields because this is the starting closure
12:19:05.478 [main] DEBUG org.apache.spark.util.ClosureCleaner -  + fields accessed by starting closure: 1
12:19:05.479 [main] DEBUG org.apache.spark.util.ClosureCleaner -      (interface org.apache.spark.streaming.api.java.JavaDStreamLike,Set())
12:19:05.479 [main] DEBUG org.apache.spark.util.ClosureCleaner -  + outermost object is not a closure, so do not clone it: (interface org.apache.spark.streaming.api.java.JavaDStreamLike,org.apache.spark.streaming.api.java.JavaDStream@7209ffb5)
12:19:05.480 [main] DEBUG org.apache.spark.util.ClosureCleaner -  +++ closure <function1> (org.apache.spark.streaming.api.java.JavaDStreamLike$$anonfun$foreachRDD$3) is now cleaned +++
12:19:05.481 [main] DEBUG org.apache.spark.util.ClosureCleaner - +++ Cleaning closure <function2> (org.apache.spark.streaming.dstream.DStream$$anonfun$foreachRDD$1$$anonfun$apply$mcV$sp$3) +++
12:19:05.482 [main] DEBUG org.apache.spark.util.ClosureCleaner -  + declared fields: 2
12:19:05.482 [main] DEBUG org.apache.spark.util.ClosureCleaner -      public static final long org.apache.spark.streaming.dstream.DStream$$anonfun$foreachRDD$1$$anonfun$apply$mcV$sp$3.serialVersionUID
12:19:05.482 [main] DEBUG org.apache.spark.util.ClosureCleaner -      private final scala.Function1 org.apache.spark.streaming.dstream.DStream$$anonfun$foreachRDD$1$$anonfun$apply$mcV$sp$3.cleanedF$1
12:19:05.482 [main] DEBUG org.apache.spark.util.ClosureCleaner -  + declared methods: 2
12:19:05.482 [main] DEBUG org.apache.spark.util.ClosureCleaner -      public final java.lang.Object org.apache.spark.streaming.dstream.DStream$$anonfun$foreachRDD$1$$anonfun$apply$mcV$sp$3.apply(java.lang.Object,java.lang.Object)
12:19:05.482 [main] DEBUG org.apache.spark.util.ClosureCleaner -      public final void org.apache.spark.streaming.dstream.DStream$$anonfun$foreachRDD$1$$anonfun$apply$mcV$sp$3.apply(org.apache.spark.rdd.RDD,org.apache.spark.streaming.Time)
12:19:05.482 [main] DEBUG org.apache.spark.util.ClosureCleaner -  + inner classes: 0
12:19:05.482 [main] DEBUG org.apache.spark.util.ClosureCleaner -  + outer classes: 0
12:19:05.482 [main] DEBUG org.apache.spark.util.ClosureCleaner -  + outer objects: 0
12:19:05.482 [main] DEBUG org.apache.spark.util.ClosureCleaner -  + populating accessed fields because this is the starting closure
12:19:05.483 [main] DEBUG org.apache.spark.util.ClosureCleaner -  + fields accessed by starting closure: 0
12:19:05.483 [main] DEBUG org.apache.spark.util.ClosureCleaner -  + there are no enclosing objects!
12:19:05.483 [main] DEBUG org.apache.spark.util.ClosureCleaner -  +++ closure <function2> (org.apache.spark.streaming.dstream.DStream$$anonfun$foreachRDD$1$$anonfun$apply$mcV$sp$3) is now cleaned +++

Am I missing something? PS: given output were just in place where my printed list should be, and i'm using Spring Unit Tests for my work:

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(classes = config.class)
public class myTester extends SharedJavaSparkContext implements Serializable{

Answer 1

I guess you need to start the streaming context.

streamingContext.start()