Я использую данные от Kafka, которые я могу успешно сделать, но теперь я хочу выполнить некоторую операцию над ним, сохранив данные в фрейме данных. Я новичок в Кафке и не знаю, как это сделать
config <- spark_config()
config$sparklyr.shell.packages <- "org.apache.spark:spark-sql-kafka-0-10_2.11:2.4.0"
sc <- spark_connect(master = "local", config = config)
read_options <- list(kafka.bootstrap.servers = kafka-intelligeni-cp-kafka.intelligeni-2:9092, subscribe = "metrics")
write_options <- list(kafka.bootstrap.servers = kafka-intelligeni-cp-kafka.intelligeni-2:9092, topic = "topics2")
stream <- stream_read_kafka(sc, options = read_options) %>% stream_write_kafka(options = write_options,trigger = stream_trigger_interval(20.0))
stream_view(stream)
stream_stop(stream)