Kafka Direct Stream в DataFrame не работает с окном - PullRequest
Новая
       10

Kafka Direct Stream в DataFrame не работает с окном

0 голосов
/ 16 января 2019

У меня есть данный кусок кода. Я запускаю это на Spark 2.3.1, используя Zeppelin 0.8.0 и org.apache.spark: spark-streaming-kafka-0-10_2.11: 2.3.1. 

stream.window(Minutes(5),Seconds(20)).foreachRDD { rdd =>
    val lines = rdd.map(record => record.value())
    val words = lines.flatMap(line => line.split(" "))
    val pairs = words.map(word => (word, 1))
    val wordCounts = pairs.reduceByKey((x: Int, y: Int) => (x + y))
    wordCounts.toDF("word", "count").createOrReplaceTempView("words")
}

Однако, когда я пытаюсь запросить таблицу слов внутри скользящего окна, я получаю следующую ошибку: 

org.apache.spark.SparkException: Job aborted due to stage failure: Task 0.0 in stage 540.0 (TID 3036) had a not serializable result: org.apache.kafka.clients.consumer.ConsumerRecord
Serialization stack:
    - object not serializable (class: org.apache.kafka.clients.consumer.ConsumerRecord, value: ConsumerRecord(topic = test, partition = 0, offset = 249, CreateTime = 1547626717449, checksum = 3583250337, serialized key size = -1, serialized value size = 4, key = null, value = test))

Любые предложения о том, как заставить это работать? Похоже, очень простой пример.

Если я запускаю его без оконной функции, он работает правильно.

1 Ответ

0 голосов
/ 16 января 2019

Очевидно, вам нужно преобразовать поток в преобразовании (...) в поток без ConsumerRecord. Затем вы можете вызвать окно на очищенный поток. Затем вы можете преобразовать этот поток и построить результирующую таблицу. 

val cleanedStream = kafkaStream.transform(rdd => rdd.map(record => record.value))

val windowedStream = cleanedStream.window(Minutes(5),Seconds(20))

val transformedStream = windowedStream.transform(rdd => {
    val words = rdd.flatMap(line => line.split(" "))
    val pairs = words.map(word => (word, 1))
    pairs.reduceByKey((x: Int, y: Int) => (x + y))
})

transformedStream.foreachRDD { rdd =>
    rdd.toDF("word", "count").createOrReplaceTempView("words")
}
...