A few years ago a colleague of mine told me about this thing called Docker, and I must admit I dismissed it as a fad…how wrong was I. Docker, and Docker Compose, are one of my key tools of the trade. With them I can build self-contained environments for tutorials, demos, conference talks etc. Tear it down, run it again, without worrying that somewhere a local config changed and will break things.

Continue Reading

This is a short summary discussing what the options are for integrating Oracle RDBMS into Kafka, as of December 2018. For a more detailed background to why and how at a broader level for all databases (not just Oracle) see this blog and these slides. What techniques & tools are there? As of December 2018, this is what the line-up looks like: Query-based CDC The JDBC Connector for Kafka Connect, polls the database for new or changed data based on an incrementing ID column and/or update timestamp Log-based CDC

Continue Reading

The problem - nested messages in Kafka Data comes into Kafka in many shapes and sizes. Sometimes it’s from CDC tools, and may be nested like this: { "SCN": 12206116841348, "SEG_OWNER": "KFKUSER", "TABLE_NAME": "CDCTAB2", "TIMESTAMP": 1539162785000, "SQL_REDO": "insert into \"KFKUSER\".\"CDCTAB2\"(\"ID\",\"CITY\",\"NATIONALITY\") values (634789,'AHMEDABAD','INDIA')", "OPERATION": "INSERT", "data": { "value": { "ID": 634789, "CITY": { "string": "AHMEDABAD" }, "NATIONALITY": { "string": "INDIA" } } }, "before": null } Note that the ‘payload’ is nested under data->value.

Continue Reading

Check out the jmxterm repository / Download jmxterm from http://wiki.cyclopsgroup.org/jmxterm/ Launch: java -jar ~/Downloads/jmxterm-1.0.0-uber.jar --url localhost:30002 You can pass the jmx host/port directly, or use the open command once jmxterm launches. Once connected, use domains to list available domains $>domains #following domains are available JMImplementation com.sun.management io.confluent.ksql.metrics io.confluent.rest java.lang java.nio java.util.logging kafka.admin.client kafka.consumer kafka.producer kafka.streams [...] Switch to a particular domain: $>domain io.confluent.ksql.metrics #domain is set to io.confluent.ksql.metrics List the available MBeans in a the selected domain (you can also run this without choosing a domain first, to see every MBean, but it’s a long list):

Continue Reading

KSQL provides the ability to create windowed aggregations. For example, count the number of messages in a 1 minute window, grouped by a particular column: CREATE TABLE RATINGS_BY_CLUB_STATUS AS \ SELECT CLUB_STATUS, COUNT(*) AS RATING_COUNT \ FROM RATINGS_WITH_CUSTOMER_DATA \ WINDOW TUMBLING (SIZE 1 MINUTES) \ GROUP BY CLUB_STATUS; How KSQL, and Kafka Streams, stores the window timestamp associated with an aggregate, has recently changed. See #1497 for details. Whereas previously the Kafka message timestamp (accessible through the KSQL ROWTIME system column) stored the start of the window for which the aggregate had been calculated, this changed in July 2018 to instead be the timestamp of the latest message to update that aggregate value.

Continue Reading

In this article I demonstrated how to use KSQL to filter streams of network event data. As well as filtering, KSQL can be used to easily enrich streams. In this article we’ll see how this enriched data can be used to drive analysis in Elasticsearch and Kibana—and how KSQL again came into use for building some stream processing as a result of the discovery made. The data came from my home Ubiquiti router, and took two forms:

Continue Reading

(preserving this StackOverflow answer for posterity and future Googlers) tl;dr When doing a stream-table join, your table messages must already exist (and must be timestamped) before the stream messages. If you re-emit your source stream messages, after the table topic is populated, the join will succeed. Example data Use kafakcat to populate topics: kafkacat -b localhost:9092 -P -t sessionDetails <<EOF {"Media":"Foo","SessionIdTime":"2018-05-17 11:25:33 BST","SessionIdSeq":1} {"Media":"Foo","SessionIdTime":"2018-05-17 11:26:33 BST","SessionIdSeq":2} EOF kafkacat -b localhost:9092 -P -t voipDetails <<EOF {"SessionIdTime":"2018-05-17 11:25:33 BST","SessionIdSeq":1,"Details":"Bar1a"} {"SessionIdTime":"2018-05-17 11:25:33 BST","SessionIdSeq":1,"Details":"Bar1b"} {"SessionIdTime":"2018-05-17 11:26:33 BST","SessionIdSeq":2,"Details":"Bar2"} EOF Validate topic contents:

Continue Reading