# Read via REST API from Kafka topic # Prerequisite: set up kafka with topic accesslogapache library(httr) library(jsonlite) library(dplyr) kafka_rest_proxy <- "http://localhost:8082" consumer <- 'my_json_consumer' consumer_instance <- 'my_consumer_instance' topic <- 'accesslogapache' # Create consumer # See https://docs.confluent.io/current/kafka-rest/docs/intro.html#produce-and-consume-avro-messages # Create consumer source("R/CreateKafkaConsumer.R") consumerDetails <- CreateKafkaConsumer(kafka.rest.proxy = kafka_rest_proxy, consumer = consumer, consumer_instance = consumer_instance) # Subscribe it to topic response <- POST(url=paste(consumerDetails$base_uri, "subscription", sep="/"), content_type("application/vnd.kafka.v2+json"), body = paste0('{"topics":["', topic, '"]}') ) response # Obtain all (or latest) messages on the topic source("R/ConsumeKafkaMessages.R") messages <- ConsumeKafkaMessages(consumerDetails$base_uri) createPlot(messages$value) Sys.sleep(120) # Obtain latest messages messagesJSON <- GET(url = paste(consumerDetails$base_uri,"records", sep = "/"), accept("application/vnd.kafka.json.v2+json"), encode="json") messages2 <- fromJSON(content(messagesJSON,"text")) apachelog <- rbind(messages$value,messages2$value) createPlot(apachelog) # Remove the consumer source("R/DestroyKafkaConsumer.R") response <- DestroyKafkaConsumer(consumer.base.uri = consumerDetails$base_uri)