// Automatically provides ssc: JavaStreamingContext which starts and awaits termination or timeout
withSparkStreaming(batchDuration = Durations.seconds(1), timeout = 10_000) { // this: KSparkStreamingSession
    setRunAfterStart {
        println("Stream is started")
    }
    // create input stream for, for instance, Netcat: `$ nc -lk 9999`
    val lines: JavaReceiverInputDStream<String> = ssc.socketTextStream("localhost", 9999)
  
    // split input stream on space
    val words: JavaDStream<String> = lines.flatMap { it.split(" ").iterator() }
    // perform action on each formed RDD in the stream
    words.foreachRDD { rdd: JavaRDD<String>, _: Time ->
      
          // to convert the JavaRDD to a Dataset, we need a spark session using the RDD context
          withSpark(rdd) { // this: KSparkSession
            val dataframe: Dataset<TestRow> = rdd.map { TestRow(word = it) }.toDS()
            dataframe
                .groupByKey { it.word }
                .count()
                .show()
            // +-----+--------+
            // |  key|count(1)|
            // +-----+--------+
            // |hello|       1|
            // |   is|       1|
            // |    a|       1|
            // | this|       1|
            // | test|       3|
            // +-----+--------+
        }
    }
}

withSparkStreaming(...) { // this: KSparkStreamingSession
    // for instance
    val broadcastSomething: Broadcast<*> = withSpark(sscForConf = ssc) { // this: KsparkSession
        spark.broadcast(something)
    }
}

withSparkStreaming(...) { // this: KSparkStreamingSession
    
    val stream: JavaDStream<*> = ...
    stream.foreachRDD { rdd: JavaRDD<*>, time: Time ->
        
        // for instance
        withSpark(rddForConf = rdd) { // this: KSparkSession
            rdd.toDS().show()
        }
    }
}

withSparkStreaming(batchDuration = ..., checkPointPath = "/path/to/checkpoint") {
    // contents will only only be run the first time
    // the second time, the stream transformations will be read from the checkpoint
}

val wordCounts: JavaDStream<Tuple2<String, Int>> = words
    .map { it X 1 }
    .reduceByKey { a: Int, b: Int -> a + b }