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