Обработка списка в DStream

Question

У меня есть список слов как DStream. Например: Список (автомобиль, скорость, авария, скорость, плохо). Я хочу сформировать bi грамм из этого списка. У меня это с RDD, но с проблемами с DStreams. Я использую функцию foreachRDD. Ниже приведено то, что у меня есть -

Я пытаюсь распечатать содержимое RDD после преобразования.

def printRDD(rddString: RDD[String]) ={ 
     val z = rddString.map(y => y.toString.split(",").filter(_.nonEmpty). 
     map(y => y.replaceAll("""\W""", "").toLowerCase) 
     .filter(_.nonEmpty) 
     .sliding(2).filter(_.size == 2).map{ case Array(a, b) => ((a, b), 1) }) 
     .flatMap(x => x) 
     println(z) 
} 
val x = lines.map(plainTextToLemmas(_, stopWords)) 
val words = x.flatMap(y=> y.toString.split(",")) 
words.foreachRDD(rdd => printRDD(rdd)) 

Есть ли способ показать содержимое после функции преобразования printRDD. Даже если я использую println (z) внутри определения печати, он возвращает MapPartitionsRDD [18] в flatMap. Я использую искрообразование Kafka для чтения входов, я получаю значение слова на консоли. Я думаю, что слова не меняются после вызова функции printRDD.

Answer 1

Вы можете сделать все это работает на DStream, а не внутри foreachRDD, а затем вызвать print на DStream:

lines 
    .map(plainTextToLemmas(_, stopWords)) 
    .flatMap(y => y.toString.split(",")) 
    .map(y => y.toString.split(",").filter(_.nonEmpty)) 
    .map(y => y.replaceAll("""\W""", "").toLowerCase) 
    .filter(_.nonEmpty) 
    .sliding(2) 
    .filter(_.size == 2) 
    .flatMap { case Array(a, b) => ((a, b), 1) } 
    .print() 

Это необходимо распечатать содержание DStream к консоли на драйвере.

Важная вещь, чтобы отметить, что, хотя вы работаете на DStream, это методы «сверла в» лежащий в основе RDD в данный момент времени периодического и выставить фактический тип внутри RDD, так что вам не нужно используйте foreachRDD, чтобы получить фактические данные внутри.