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 response <- POST(url=paste(consumerDetails$base_uri,
  23                            "subscription", sep="/"),
  24                content_type("application/vnd.kafka.v2+json"),
  25                body = paste0('{"topics":["',
  26                              topic,
  27                              '"]}')
  28                )
  29 response
  30
  31 # Obtain all (or latest) messages on the topic
  32 source("R/ConsumeKafkaMessages.R")
  33 messages <- ConsumeKafkaMessages(consumerDetails$base_uri)
  34
  35 createPlot(messages$value)
  36
  37 Sys.sleep(120)
  38
  39 # Obtain latest messages
  40 messagesJSON <- GET(url = paste(consumerDetails$base_uri,"records", sep = "/"),
  41                 accept("application/vnd.kafka.json.v2+json"),
  42                 encode="json")
  43
  44 messages2 <- fromJSON(content(messagesJSON,"text"))
  45
  46 apachelog <- rbind(messages$value,messages2$value)
  47
  48 createPlot(apachelog)
  49
  50 # Remove the consumer
  51 source("R/DestroyKafkaConsumer.R")
  52 response <- DestroyKafkaConsumer(consumer.base.uri = consumerDetails$base_uri)