MapR Ecosystem Package 2.0 (MEP) is coming with some new features related to MapR Streams:<\/p>\n
MapR Ecosystem Packs (MEPs) are a way to deliver ecosystem upgrades decoupled from core upgrades – allowing you to upgrade your tooling independently of your Converged Data Platform. You can lean more about MEP 2.0 in this article<\/a>.<\/p>\n In this blog we describe how to use the REST Proxy to publish and consume messages to\/from MapR Streams. The REST Proxy is a great addition to the MapR Converged Data Platform allowing any programming language to use MapR Streams.<\/p>\n The Kafka REST Proxy provided with the MapR Streams tools, can be used with MapR Streams (default), but also used in a hybrid mode with Apache Kafka. In this article we will focus on MapR Streams. <!– more –><\/p>\n A stream is a collection of topics that you can manage as a group by:<\/p>\n You can find more information about MapR Streams concepts in the documentation<\/a>.<\/p>\n On your Mapr Cluster or Sandbox, run the following commands:<\/p>\n Open two terminal windows and run the consumer Kafka utilities using the following commands:<\/p>\n This two terminal windows will allow you to see the messages posted on the different topics<\/p>\n The endpoint Run the following command, to get information about the Note: For simplicity reason I am running the command from the node where the Kafka REST proxy is running, so it is possible to use You can print JSON in a pretty way, by adding a Python command such as :<\/p>\n Default Stream<\/strong><\/p>\n As mentioned above, the Stream path is part of the topic name you have to use in the command; however it is possible to configure the MapR Kafka REST Proxy to use a default stream. For this you should add the following property in the In this article, all the URLs contains the encoded stream name, like that you can start using the Kafka REST proxy without changind the configuration and also use it with different streams.<\/li>\n<\/ul>\n The Kafka REST Proxy for MapR Streams allows application to publish messages to MapR Streams. Messages could be send as JSON or Binary content (base64 encoding).<\/p>\n To send a JSON Message:<\/strong><\/p>\n The complete request is:<\/p>\n You should see the message printed in the terminal window where the To send a binary Message:<\/strong><\/p>\n Note that The complete request is:<\/p>\n You should see the message printed in the terminal window where the Sending multiple messages<\/strong><\/p>\n The This command will send 2 messages, and increment the offset by 2. You can do the same with binary content, just add new element in the JSON array; for example:<\/p>\n As you probably know, it is possible to set a key to a message to be sure that all the messages with the same key will arrive in the same partition. For this, add the Now that you know how to post messages to MapR Stream topics usinf the REST Proxy, let’s see how to consume the messages.<\/p>\n The REST proxy can also be used to consume messages from topics; for this you need to:<\/p>\n Creating the consumer instance<\/strong><\/p>\n The following request creates the consumer instance:<\/p>\n The response from the server looks like:<\/p>\n Note that we have used the The Consuming the messages<\/strong><\/p>\n To consume the messages, just add the Mapr Streams topic to the URL of the consumer isntance.<\/p>\n The following request consumes the messages from the topic:<\/p>\n This call returns the messages in a JSON document:<\/p>\n Each call to the API returns the new messages published, based on the offset of the last call.<\/p>\n Note that the Consumer will be destroyed:<\/p>\n The approach is the same if you need to consume binary messages, you need to change the format and accept header.<\/p>\n Call this URL to create a consumer instance for the binary topic:<\/p>\n Then consume messages, the accept header is set to This call returns the messages in a JSON document, and the value is encoded in Base64<\/p>\n As mentioned before the consumer will be destroyed automatically based on the In this article you have learned how to use the Kafka REST Proxy for MapR Streams that allow any application to use messages published in the MapR Converged Data Platform.<\/p>\n You can find more information about the Kafka REST Proxy in the MapR documentation<\/a> and the following resources:<\/p>\nPrerequisites<\/h2>\n
\n
\n
Create the MapR Streams and Topic<\/h2>\n
\n
$ maprcli stream create -path \/apps\/iot-stream -produceperm p -consumeperm p -topicperm p\r\n\r\n$ maprcli stream topic create -path \/apps\/iot-stream -topic sensor-json -partitions 3\r\n\r\n$ maprcli stream topic create -path \/apps\/iot-stream -topic sensor-binary -partitions 3<\/pre>\n
Start Kafka Console Producers and Consumers<\/h2>\n
Consumer<\/h3>\n
\n
$ \/opt\/mapr\/kafka\/kafka-0.9.0\/bin\/kafka-console-consumer.sh --new-consumer --bootstrap-server this.will.be.ignored:9092 --topic \/apps\/iot-stream:sensor-json<\/pre>\n
\n
$ \/opt\/mapr\/kafka\/kafka-0.9.0\/bin\/kafka-console-consumer.sh --new-consumer --bootstrap-server this.will.be.ignored:9092 --topic \/apps\/iot-stream:sensor-binary<\/pre>\n
Using Kafka REST Proxy<\/h2>\n
Inspect Topic Metadata<\/h3>\n
\/topics\/[topic_name]<\/code> allows you to get some informations about the topic. In MapR Streams, topics are part of a stream<\/em> identified by a path; to use the topic using the REST API you have to use the full path, and encode it in the URL; for example:<\/p>\n\n
\/apps\/iot-stream:sensor-json<\/code> will be encoded with %2Fapps%2Fiot-stream%3Asensor-json<\/code><\/li>\n<\/ul>\nsensor-json<\/code> topic<\/p>\n$ curl -X GET http:\/\/localhost:8082\/topics\/%2Fapps%2Fiot-stream%3Asensor-json<\/pre>\n
localhost<\/code>.<\/p>\n$ curl -X GET http:\/\/localhost:8082\/topics\/%2Fapps%2Fiot-stream%3Asensor-json | python -m json.tool<\/pre>\n
\/opt\/mapr\/kafka-rest\/kafka-rest-2.0.1\/config\/kafka-rest.properties<\/code> file:\n
streams.default.stream=\/apps\/iot-stream<\/code>When you change the Kafka REST proxy configuration, you must restart the service using maprcli or MCS.The main reason to use the streams.default.stream<\/code> properties is to simplify the URLs used by the application for example\n\n
streams.default.stream<\/code> you can use curl -X GET http:\/\/localhost:8082\/topics\/<\/code><\/li>\nhttp:\/\/localhost:8082\/topics\/%2Fapps%2Fiot-stream%3Asensor-json<\/code><\/li>\n<\/ul>\nPublishing Messages<\/h3>\n
\n
POST<\/code><\/li>\napplication\/vnd.kafka.json.v1+json<\/code><\/li>\n{\r\n \"records\":\r\n [\r\n {\r\n \"value\":\r\n {\r\n \"temp\" : 10 ,\r\n \"speed\" : 40 ,\r\n \"direction\" : \"NW\"\r\n } \r\n }\r\n ]\r\n}<\/pre>\ncurl -X POST -H \"Content-Type: application\/vnd.kafka.json.v1+json\" \\\r\n --data '{\"records\":[{\"value\": {\"temp\" : 10 , \"speed\" : 40 , \"direction\" : \"NW\"} }]}' \\\r\n http:\/\/localhost:8082\/topics\/%2Fapps%2Fiot-stream%3Asensor-json<\/pre>\n\/apps\/iot-stream:sensor-json<\/code> consumer is running.<\/p>\n\n
POST<\/code><\/li>\napplication\/vnd.kafka.binary.v1+json<\/code><\/li>\n{\r\n \"records\":\r\n [\r\n {\r\n \"value\":\"SGVsbG8gV29ybGQ=\"\r\n }\r\n ]\r\n}<\/pre>\nSGVsbG8gV29ybGQ=<\/code> is the string “Hello World” encoded in Base64.<\/p>\ncurl -X POST -H \"Content-Type: application\/vnd.kafka.binary.v1+json\" \\\r\n --data '{\"records\":[{\"value\":\"SGVsbG8gV29ybGQ=\"}]}' \\\r\n http:\/\/localhost:8082\/topics\/%2Fapps%2Fiot-stream%3Asensor-binary<\/pre>\n\/apps\/iot-stream:sensor-binary<\/code> consumer is running.<\/p>\nrecords<\/code> field of the HTTP Body allows you to send multiple messages for example you can send:<\/p>\ncurl -X POST -H \"Content-Type: application\/vnd.kafka.json.v1+json\" \\\r\n --data '{\"records\":[{\"value\": {\"temp\" : 12 , \"speed\" : 42 , \"direction\" : \"NW\"} }, {\"value\": {\"temp\" : 10 , \"speed\" : 37 , \"direction\" : \"N\"} } ]}' \\\r\n http:\/\/localhost:8082\/topics\/%2Fapps%2Fiot-stream%3Asensor-json<\/pre>\ncurl -X POST -H \"Content-Type: application\/vnd.kafka.binary.v1+json\" \\\r\n --data '{\"records\":[{\"value\":\"SGVsbG8gV29ybGQ=\"}, {\"value\":\"Qm9uam91cg==\"}]}' \\\r\n http:\/\/localhost:8082\/topics\/%2Fapps%2Fiot-stream%3Asensor-binary<\/pre>\nkey<\/code> attribute to the message as follow:<\/p>\n{\r\n \"records\":\r\n [\r\n {\r\n \"key\": \"K001\",\r\n \"value\":\r\n {\r\n \"temp\" : 10 ,\r\n \"speed\" : 40 ,\r\n \"direction\" : \"NW\"\r\n } \r\n }\r\n ]\r\n}<\/pre>\nConsuming Messages<\/h3>\n
\n
curl -X POST -H \"Content-Type: application\/vnd.kafka.v1+json\" \\\r\n --data '{\"name\": \"iot_json_consumer\", \"format\": \"json\", \"auto.offset.reset\": \"earliest\"}' \\\r\n http:\/\/localhost:8082\/consumers\/%2Fapps%2Fiot-stream%3Asensor-json<\/pre>\n{\r\n \"instance_id\":\"iot_json_consumer\",\r\n \"base_uri\":\"http:\/\/localhost:8082\/consumers\/%2Fapps%2Fiot-stream%3Asensor-json\/instances\/iot_json_consumer\"\r\n}<\/pre>\n\/consumers\/[topic_name]<\/code> to create the consumer.<\/p>\nbase_uri<\/code> will be used by the subsequent requests to get the messages from the topic. Like any MapR Streams\/Kafka consumer the auto.offset.reset<\/code> defines its behavior. In this example the value is set to earliest<\/code>, this means that the consumer will read the messages from the beginning. You can find more information about the consumer configuration in the MapR Streams documentation<\/a>.<\/p>\ncurl -X GET -H \"Accept: application\/vnd.kafka.json.v1+json\" \\\r\nhttp:\/\/localhost:8082\/consumers\/%2Fapps%2Fiot-stream%3Asensor-json\/instances\/iot_json_consumer\/topics\/%2Fapps%2Fiot-stream%3Asensor-json<\/pre>\n
[\r\n {\"key\":null,\"value\":{\"temp\":10,\"speed\":40,\"direction\":\"NW\"},\"topic\":\"\/apps\/iot-stream:sensor-json\",\"partition\":1,\"offset\":1},\r\n {\"key\":null,\"value\":{\"temp\":12,\"speed\":42,\"direction\":\"NW\"},\"topic\":\"\/apps\/iot-stream:sensor-json\",\"partition\":1,\"offset\":2},\r\n {\"key\":null,\"value\":{\"temp\":10,\"speed\":37,\"direction\":\"N\"},\"topic\":\"\/apps\/iot-stream:sensor-json\",\"partition\":1,\"offset\":3}\r\n]<\/pre>\n\n
consumer.instance.timeout.ms<\/code> (default value set to 300000ms \/ 5 minutes)<\/li>\nConsuming binary format messages<\/h3>\n
curl -X POST -H \"Content-Type: application\/vnd.kafka.v1+json\" \\\r\n --data '{\"name\": \"iot_binary_consumer\", \"format\": \"binary\", \"auto.offset.reset\": \"earliest\"}' \\\r\n http:\/\/localhost:8082\/consumers\/%2Fapps%2Fiot-stream%3Asensor-binary<\/pre>\napplication\/vnd.kafka.binary.v1+json<\/code>:<\/p>\ncurl -X GET -H \"Accept: application\/vnd.kafka.binary.v1+json\" \\\r\nhttp:\/\/localhost:8082\/consumers\/%2Fapps%2Fiot-stream%3Asensor-binary\/instances\/iot_binary_consumer\/topics\/%2Fapps%2Fiot-stream%3Asensor-binary<\/pre>\n
[\r\n {\"key\":null,\"value\":\"SGVsbG8gV29ybGQ=\",\"topic\":\"\/apps\/iot-stream:sensor-binary\",\"partition\":1,\"offset\":1},\r\n {\"key\":null,\"value\":\"Qm9uam91cg==\",\"topic\":\"\/apps\/iot-stream:sensor-binary\",\"partition\":1,\"offset\":2}\r\n]<\/pre>\nDelete consumer instances<\/h3>\n
consumer.instance.timeout.ms<\/code> configuration of the REST Proxy; it is also possible to destroyed the instance using the consumer instance URI and an HTTP DELETE call, as follow:<\/p>\ncurl -X DELETE http:\/\/localhost:8082\/consumers\/%2Fapps%2Fiot-stream%3Asensor-binary\/instances\/iot_binary_consumer<\/pre>\n
Conclusion<\/h2>\n
\n