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 messagesJSON <- GET(url = paste(consumerDetails$base_uri,"records", sep = "/"),
  33                 accept("application/vnd.kafka.json.v2+json"),
  34                 encode="json")
  35
  36 messages <- fromJSON(content(messagesJSON,"text"))
  37
  38 createPlot(messages$value)
  39
  40 Sys.sleep(120)
  41
  42 # Obtain latest messages
  43 messagesJSON <- GET(url = paste(consumerDetails$base_uri,"records", sep = "/"),
  44                 accept("application/vnd.kafka.json.v2+json"),
  45                 encode="json")
  46
  47 messages2 <- fromJSON(content(messagesJSON,"text"))
  48
  49 apachelog <- rbind(messages$value,messages2$value)
  50
  51 createPlot(apachelog)
  52
  53 # Remove the consumer
  54 DELETE(url = consumerDetails$base_uri,
  55        content_type("application/vnd.kafka.v2+json"))