kafkaConsumer.R

   1 # Read via REST API from Kafka topic
   2 # Prerequisite: set up kafka with topic accesslogapache
   3
   4
   5 library(httr)
   6 library(jsonlite)
   7 library(dplyr)
   8
   9 kafka_rest_proxy <- "http://localhost:8082"
  10 consumer <- 'my_json_consumer'
  11 consumer_instance <- 'my_consumer_instance'
  12 topic <- 'accesslogapache'
  13
  14 # Create consumer
  15 # See https://docs.confluent.io/current/kafka-rest/docs/intro.html#produce-and-consume-avro-messages
  16
  17 # Create consumer
  18 source("R/CreateKafkaConsumer.R")
  19 consumerDetails <- CreateKafkaConsumer(kafka.rest.proxy = kafka_rest_proxy, consumer = consumer, consumer_instance = consumer_instance)
  20
  21 # Subscribe it to topic
  22 source("R/SubscribeKafkaTopic.R")
  23 response <- SubscribeKafkaTopic(consumerDetails$base_uri,topic = topic)
  24
  25 # Obtain all (or latest) messages on the topic
  26 source("R/ConsumeKafkaMessages.R")
  27 messages <- ConsumeKafkaMessages(consumerDetails$base_uri)
  28
  29 createPlot(messages$value)
  30
  31 Sys.sleep(120)
  32
  33 # Obtain latest messages
  34 messagesJSON <- GET(url = paste(consumerDetails$base_uri,"records", sep = "/"),
  35                 accept("application/vnd.kafka.json.v2+json"),
  36                 encode="json")
  37
  38 messages2 <- fromJSON(content(messagesJSON,"text"))
  39
  40 apachelog <- rbind(messages$value,messages2$value)
  41
  42 createPlot(apachelog)
  43
  44 # Remove the consumer
  45 source("R/DestroyKafkaConsumer.R")
  46 response <- DestroyKafkaConsumer(consumer.base.uri = consumerDetails$base_uri)