Annnnd... it's NOT working. Here's the code:
val bytes = kafkaStream.map({ case (key, messageBytes) => messageBytes}) // Map to just get the bytes part out... val things = bytes.flatMap(bytesArrayToThings) // convert to a thing val srcDestinations = things.map(thing => (ipToString(thing.getSourceIp), Set(ipToString(thing.getDestinationIp)))) // up to this point works fine. // this fails to print val srcDestinationSets = srcDestinations.reduceByKey((exist: Set[String], addl: Set[String]) => exist ++ addl) What it does... >From a kafka message containing many "things", convert the message to an array of said "things", flatMap them out to a stream of 1 "thing" at a time, pull out and make a Tuple of a (SourceIP, DestinationIP). ALL THAT WORKS. If I do a "srcDestinations.print()" I get output like the following, every 5 seconds, which is my batch size. ------------------------------------------- Time: 1402582000000 ms ------------------------------------------- (10.30.51.216,Set(10.20.1.1)) (10.20.11.3,Set(10.10.61.98)) (10.20.11.3,Set(10.10.61.95)) ... What I want is a SET of (sourceIP -> Set(all the destination Ips)) Using a set because as you can see above, the same source may have the same destination multiple times and I want to eliminate dupes on the destination side. When I call the reduceByKey() method, I never get any output. When I do a "srcDestinationSets.print()" NOTHING EVER PRINTS. Ever. Never. What am I doing wrong? (The same happens for "reduceByKeyAndWindow(..., Seconds(5))".) I'm sure this is something I've done, but I cannot figure out what it was. Help, please?