Flatten CDC records in KSQL
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:
Streaming geopoint data from Kafka to Elasticsearch
Using the Elasticsearch Kafka Connect connector to stream events from a Kafka topic to Elasticsearch.
Exploring JMX with jmxterm
- Check out the jmxterm repository
- Download jmxterm from https://docs.cyclopsgroup.org/jmxterm
Accessing Kafka Docker containers’ JMX from host
See also docs.
To help future Googlers… with the Confluent docker images for Kafka, KSQL, Kafka Connect, etc, if you want to access JMX metrics from within, you just need to pass two environment variables: <x>_JMX_HOSTNAME and <x>_JMX_PORT, prefixed by a component name.
<x>_JMX_HOSTNAME - the hostname/IP of the JMX host machine, as accessible from the JMX Client.
This is used by the JMX client to connect back into JMX, so must be accessible from the host machine running the JMX client.
Sending multiline messages to Kafka
(SO answer repost)
You can use kafkacat to send messages to Kafka that include line breaks. To do this, use its -D operator to specify a custom message delimiter (in this example /):
kafkacat -b kafka:29092 \ -t test_topic_01 \ -D/ \ -P <<EOF this is a string message with a line break/this is another message with two line breaks! EOF Note that the delimiter must be a single byte - multi-byte chars will end up getting included in the resulting message See issue #140
Window Timestamps in KSQL / Integration with Elasticsearch
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.
Converting from AsciiDoc to MS Word
Short and sweet this one. I’ve written in the past how I love Markdown but I’ve actually moved on from that and now firmly throw my hat in the AsciiDoc ring. I’ll write another post another time explaining why in more detail, but in short it’s just more powerful whilst still simple and readable without compilation.
So anyway, I use AsciiDoc (ADOC) for all my technical (and often non-technical) writing now, and from there usually dump it out to HTML which I can share with people as needed:
Where I’m speaking in the rest of 2018
There’s lots going on in the next few months :-)
I’m particularly excited to be speaking at several notable conferences for the first time, including JavaZone, USENIX LISA, and Devoxx.
As always, if you’re nearby then hope to see you there, and let me know if you want to meet for a coffee or beer!
September 🇪🇸 Madrid, Spain 6th Sept: Madrid Kafka Meetup 🇳🇴 Oslo, Norway 10th Sept: Oslo Kafka Meetup 11th Sept: JavaZone 🇧🇪 Antwerp/Brussels, Belgium 25th Sept: Brussels Kafka Meetup 🇪🇸 Barcelona, Spain 26th Sept: Barcelona Kafka Meetup October 🇬🇧 Leeds, UK 4th Oct: The JVM Thing meetup 🇺🇸 Nashville (TN), USA 31st Oct: LISA18 November 🇩🇪 Münich, Germany 7th Nov: W-JAX 🇧🇪 Antwerp, Belgium 13th Nov: Devoxx Belgium 🇵🇱 Krakow, Poland 26th Nov: CoreDump December 🇬🇧 Liverpool, UK 4th Dec: UKOUG TECH 18 🇩🇪 Frankfurt, Germany 10th Dec: Apache Kafka Meetup 11th Dec: IT Days
Kafka Listeners - Explained
(This was cross-posted on the Confluent.io blog)
This question comes up on StackOverflow and such places a lot, so here’s something to try and help.
tl;dr : You need to set advertised.listeners (or KAFKA_ADVERTISED_LISTENERS if you’re using Docker images) to the external address (host/IP) so that clients can correctly connect to it. Otherwise they’ll try to connect to the internal host address–and if that’s not reachable then problems ensue.
Put another way, courtesy of Spencer Ruport:
LISTENERSare what interfaces Kafka binds to.ADVERTISED_LISTENERSare how clients can connect.
Syntax highlighting code for presentation slides
So you’ve got a code sample you want to share in a presentation, but whilst it looks beautiful in your text-editor with syntax highlighting, it’s fugly in Keynote? You could screenshot it and paste the image into your slide, but you just know that you’ll want to change that code, and end up re-snapshotting it…what a PITA.
Better to have a nicely syntax-highlighted code snippet that you can paste as formatted text into Keynote and amend from there as needed.
Analysing Network Data with Apache Kafka, KSQL, and Elasticsearch
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:
Compare and apply a diff / patch recursively
Hacky way to keep config files in sync when there’s a new version of some software.
Caveat : probably completely wrong, may not pick up config entries added in the new version, etc. But, works for me right here right now ;-)
So let’s say we have two folders:
confluent-4.1.0 confluent-4.1.1 Same structures, different versions. 4.1.0 was set up with our local config in ./etc, that we want to preserve.
Kafka Connect and Oracle data types
The Kafka Connect JDBC Connector by default does not cope so well with:
NUMBER columns with no defined precision/scale. You may end up with apparent junk (bytes) in the output, or just errors. TIMESTAMP WITH LOCAL TIME ZONE. Throws JDBC type -102 not currently supported warning in the log. Read more about NUMBER data type in the Oracle docs.
tl;dr : How do I make it work? There are several options:
Stream-Table Joins in KSQL: Stream events must be timestamped after the Table messages
(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:
Quick ‘n Easy Population of Realistic Test Data into Kafka
tl;dr Use curl to pull data from the Mockaroo REST endpoint, and pipe it into kafkacat, thus:
curl -s "https://api.mockaroo.com/api/d5a195e0?count=2&key=ff7856d0"| \ kafkacat -b localhost:9092 -t purchases -P Three things I love…Kafka, kafkacat, and Mockaroo. And in this post I get to show all three 😁
Mockaroo is a very cool online service that lets you quickly mock up test data. What sets it apart from SELECT RANDOM(100) FROM DUMMY; is that it has lots of different classes of test data for you to choose from.
Streaming Data from MongoDB into Kafka with Kafka Connect and Debezium
Disclaimer: I am not a MongoDB person. These steps may or may not be appropriate and proper. But they worked for me :) Feel free to post in comments if I’m doing something wrong
MongoDB config - enabling replica sets For Debezium to be able to stream changes from MongoDB, Mongo needs to have replication configured:
Docs: Replication / Convert a Standalone to a Replica Set
Stop Mongo:
rmoff@proxmox01 ~> sudo service mongod stop Add replica set config to /etc/mongod.
Cloning Ubiquiti’s MongoDB instance to a separate server
DISCLAIMER: I am not a MongoDB person (even if it is Web Scale X-D) - below instructions may work for you, they may not. Use with care!
For some work I’ve been doing I wanted to access the data in Ubiquiti’s Unifi controller which it stores in MongoDB. Because I didn’t want to risk my actual Unifi device by changing local settings to enable remote access, and also because the version of MongoDB on it is older than ideal, I wanted to clone the data elsewhere.
Streaming Data from MySQL into Kafka with Kafka Connect and Debezium
Debezium is a CDC tool that can stream changes from MySQL, MongoDB, and PostgreSQL into Kafka, using Kafka Connect. In this article we’ll see how to set it up and examine the format of the data. A subsequent article will show using this realtime stream of data from a RDBMS and join it to data originating from other sources, using KSQL.
The software versions used here are:
Confluent Platform 4.
KSQL: Topic … does not conform to the requirements
io.confluent.ksql.exception.KafkaTopicException: Topic 'KSQL_NOTIFY' does not conform to the requirements Partitions:1 v 4. Replication: 1 v 1 Why? Because the topic KSQL creates to underpin a CREATE STREAM AS SELECT or CREATE TABLE AS SELECT already exists, and doesn’t match what it expects. By default it will create partitions & replicas based on the same values of the input topic.
Options:
Use a different topic, via the WITH (KAFKA_TOPIC='FOO') syntax, e.
Streaming data from Kafka into Elasticsearch
This article is part of a series exploring Streaming ETL in practice. You can read about setting up the ingest of realtime events from a standard Oracle platform, and building streaming ETL using KSQL.
This post shows how we take data streaming in from an Oracle transactional system into Kafka, and simply stream it onwards into Elasticsearch. This is a common pattern, for enabling rapid search or analytics against data held in systems elsewhere.