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 response <- POST(url=paste(kafka_rest_proxy, "consumers", consumer, sep="/"),
  19      content_type("application/vnd.kafka.v2+json"),
  20      accept("application/vnd.kafka.v2+json"),
  21      body=paste0('{"name": "',
  22                  consumer_instance,
  23                  '", "format": "json", "auto.offset.reset": "earliest"}')
  24      )
  25 fromJSON(content(response, "text"))
  26
  27 # Subscribe it to topic
  28 response <- POST(url=paste(kafka_rest_proxy,
  29                            "consumers",
  30                            consumer,
  31                            "instances",
  32                            consumer_instance,
  33                            "subscription", sep="/"),
  34                content_type("application/vnd.kafka.v2+json"),
  35                body = paste0('{"topics":["',
  36                              topic,
  37                              '"]}')
  38                )
  39 response
  40
  41 # Obtain all (or latest) messages on the topic
  42 messagesJSON <- GET(url = paste(kafka_rest_proxy,"consumers",consumer,"instances",consumer_instance,"records", sep = "/"),
  43                 accept("application/vnd.kafka.json.v2+json"),
  44                 encode="json")
  45
  46 messages <- fromJSON(content(messagesJSON,"text"))
  47
  48 createPlot(messages$value)
  49
  50 Sys.sleep(120)
  51
  52 # Obtain latest messages
  53 messagesJSON <- GET(url = paste(kafka_rest_proxy,"consumers",consumer,"instances",consumer_instance,"records", sep = "/"),
  54                 accept("application/vnd.kafka.json.v2+json"),
  55                 encode="json")
  56
  57 messages2 <- fromJSON(content(messagesJSON,"text"))
  58
  59 apachelog <- rbind(messages$value,messages2$value)
  60
  61 createPlot(apachelog)
  62
  63 # Remove the consumer
  64 DELETE(url = paste(kafka_rest_proxy,"consumers",consumer,"instances",consumer_instance, sep = "/"),
  65        content_type("application/vnd.kafka.v2+json"))