From: Frederik Vanrenterghem Date: Sat, 8 Sep 2018 12:24:54 +0000 (+0800) Subject: Add comments and variables for clarity. X-Git-Url: http://git.vanrenterghem.biz/R/project-using-kafka-in-R.git/commitdiff_plain/ca8babd7ae1bc42d81cd8605528aa01bd713718a?hp=37e86149af1f6d0f955c37674d27cc3075500858 Add comments and variables for clarity. --- diff --git a/kafkaConsumer.R b/kafkaConsumer.R index a2ca952..4a77beb 100644 --- a/kafkaConsumer.R +++ b/kafkaConsumer.R @@ -7,28 +7,39 @@ library(jsonlite) library(dplyr) kafka_rest_proxy <- "http://localhost:8082" +consumer <- 'my_json_consumer' +consumer_instance <- 'my_consumer_instance' +topic <- 'accesslogapache' # Create consumer # See https://docs.confluent.io/current/kafka-rest/docs/intro.html#produce-and-consume-avro-messages -response <- POST(url=paste(kafka_rest_proxy, "consumers", "my_json_consumer", sep="/"), +# Create consumer +response <- POST(url=paste(kafka_rest_proxy, "consumers", consumer, sep="/"), content_type("application/vnd.kafka.v2+json"), accept("application/vnd.kafka.v2+json"), - body='{"name": "my_consumer_instance", "format": "json", "auto.offset.reset": "earliest"}') + body=paste0('{"name": "', + consumer_instance, + '", "format": "json", "auto.offset.reset": "earliest"}') + ) fromJSON(content(response, "text")) +# Subscribe it to topic response <- POST(url=paste(kafka_rest_proxy, "consumers", - "my_json_consumer", + consumer, "instances", - "my_consumer_instance", + consumer_instance, "subscription", sep="/"), content_type("application/vnd.kafka.v2+json"), - body = '{"topics":["accesslogapache"]}') + body = paste0('{"topics":["', + topic, + '"]}') + ) response -# Obtain all messages on the topic -messagesJSON <- GET(url = paste(kafka_rest_proxy,"consumers","my_json_consumer","instances","my_consumer_instance","records", sep = "/"), +# Obtain all (or latest) messages on the topic +messagesJSON <- GET(url = paste(kafka_rest_proxy,"consumers",consumer,"instances",consumer_instance,"records", sep = "/"), accept("application/vnd.kafka.json.v2+json"), encode="json") @@ -38,7 +49,8 @@ createPlot(messages$value) Sys.sleep(120) -messagesJSON <- GET(url = paste(kafka_rest_proxy,"consumers","my_json_consumer","instances","my_consumer_instance","records", sep = "/"), +# Obtain latest messages +messagesJSON <- GET(url = paste(kafka_rest_proxy,"consumers",consumer,"instances",consumer_instance,"records", sep = "/"), accept("application/vnd.kafka.json.v2+json"), encode="json") @@ -49,5 +61,5 @@ apachelog <- rbind(messages$value,messages2$value) createPlot(apachelog) # Remove the consumer -DELETE(url = paste(kafka_rest_proxy,"consumers","my_json_consumer","instances","my_consumer_instance", sep = "/"), +DELETE(url = paste(kafka_rest_proxy,"consumers",consumer,"instances",consumer_instance, sep = "/"), content_type("application/vnd.kafka.v2+json"))