{"id":4740,"date":"2022-11-28T18:43:43","date_gmt":"2022-11-28T18:43:43","guid":{"rendered":"https:\/\/dalelane.co.uk\/blog\/?p=4740"},"modified":"2022-12-01T19:14:58","modified_gmt":"2022-12-01T19:14:58","slug":"using-ibm-datastage-to-process-json-events-on-apache-kafka-topics","status":"publish","type":"post","link":"https:\/\/dalelane.co.uk\/blog\/?p=4740","title":{"rendered":"Using IBM DataStage to process JSON events on Apache Kafka topics"},"content":{"rendered":"

In this post, I share a step-by-step guide for how to use IBM DataStage to merge JSON messages from multiple different Apache Kafka topics, into a single joined-up stream of events.<\/strong><\/p>\n

\"screenshot\"<\/a><\/p>\n

I needed to use IBM DataStage for the first time last week. As this was new to me, it feels interesting enough to share. (If nothing else, putting my notes here will probably be useful to me when I next need to remind myself how I got started!)<\/p>\n

Warning: This is the first DataStage flow I’ve created, so I’m not an authority and can’t claim this is best practice. It seems to work as far as I can tell, but it’s entirely possible I’m doing something daft (so please let me know if you see any mistakes!)<\/em><\/p>\n

For my first flow, I wanted to try merging multiple streams of events into a single joined up stream. To keep the walkthrough in this post easier to follow, I’ve kept it to just two topics, but I used this approach to combine several topics in the same way.<\/p>\n

\n

My starting point was to create a Kafka cluster using IBM Event Streams<\/a>, from Cloud Pak for Integration<\/a>.<\/p>\n

\"screenshot\"<\/a><\/p>\n

\n

I have a collection of topics for stock price updates. Each topic receives events every minute for a different company.<\/p>\n

(If you’d like to know how I get a Kafka topic with stock price events<\/a>, I’ve written about that before, but the short version is I’m using a custom Kafka Connect connector<\/a>.<\/em>)<\/p>\n

\"screenshot\"<\/a>
\n\"screenshot\"<\/a><\/p>\n

\n

Now I was ready to start processing my Kafka messages, using IBM DataStage<\/a> from Cloud Pak for Data<\/a>.<\/p>\n

First up, I clicked on New project<\/strong> to create my project.<\/p>\n

\"screenshot\"<\/a><\/p>\n

\n

I went with an empty project<\/strong> for this.<\/p>\n

\"screenshot\"<\/a><\/p>\n

\n

To start with, there are a few assets I wanted to get ready. So I clicked on to the Assets<\/strong> tab then clicked the New asset<\/strong> button.<\/p>\n

\"screenshot\"<\/a><\/p>\n

\n

The first thing to get ready was the connection to the Kafka cluster, so I chose the Connection<\/strong> asset type.<\/p>\n

\"screenshot\"<\/a><\/p>\n

\n

Lots of connection types to choose from – I wanted Apache Kafka<\/strong>.<\/p>\n

\"screenshot\"<\/a><\/p>\n

\n

I called my connection “Event Streams” because I lack imagination.<\/p>\n

\"screenshot\"<\/a><\/p>\n

\n

Next thing I needed was the bootstrap address for the Kafka cluster, which I got by clicking on the copy button for the External listener.<\/p>\n

\"screenshot\"<\/a><\/p>\n

\n

I pasted that into the Kafka server host name<\/strong> field.<\/p>\n

\"screenshot\"<\/a><\/p>\n

\n

I needed some credentials for DataStage to use, so I clicked the Generate SCRAM credentials<\/strong> button, and used the wizard to create a new username and password.<\/p>\n

\"screenshot\"<\/a><\/p>\n

\n

I pasted those into the credentials section of the DataStage connection form.<\/p>\n

\"screenshot\"<\/a><\/p>\n

\n

Finally, I needed to download the truststore certificate for the Kafka listener, which I got by clicking the download button for the PEM certificate<\/strong>.<\/p>\n

\"screenshot\"<\/a><\/p>\n

\n

I pasted the contents of that pem file into the Truststore certificates<\/strong> box on the DataStage connection form.<\/p>\n

\"screenshot\"<\/a><\/p>\n

\n

First asset ready, two more to go.<\/p>\n

Next was storing a schema that describes the JSON messages that we’ll be consuming from the Kafka topics.<\/p>\n

Back to the New asset<\/strong> button to start that.<\/p>\n

\"screenshot\"<\/a><\/p>\n

\n

Schema libraries are included in the DataStage component<\/strong> section.<\/p>\n

\"screenshot\"<\/a><\/p>\n

\n

I chose the Schema library<\/strong>.<\/p>\n

\"screenshot\"<\/a><\/p>\n

\n

This library was for the definitions of input messages – the messages on the topics like STOCK.PRICES.APPLE<\/code> and STOCK.PRICES.MICROSOFT<\/code> that I wanted to consume.<\/p>\n

\"screenshot\"<\/a><\/p>\n

\n

I clicked on Add resource<\/strong> to upload my JSON file.<\/p>\n

\"screenshot\"<\/a><\/p>\n

\n

You don’t have to upload a schema – you can just upload an example message, and DataStage will generate a schema from that.<\/p>\n

So I went back to the Event Streams web interface, and clicked on the Copy<\/strong> to copy the contents of the most recent message.<\/p>\n

\"screenshot\"<\/a><\/p>\n

My sample.json<\/code> file looked like:<\/p>\n

{\n    \"open\": 147.4,\n    \"high\": 147.58,\n    \"low\": 147.4,\n    \"close\": 147.57,\n    \"volume\": 56044,\n    \"timestamp\": 1668805140,\n    \"datetime\": \"2022-11-18 15:59:00\"\n}\n<\/pre>\n
\n
DataStage used my sample.json<\/code> file to generate a sample.json.xsd<\/strong> schema.<\/p>\n
\"screenshot\"<\/a><\/p>\n
\n
I also wanted to create a schema library for the messages I wanted my DataStage flows to output, so back to the Assets tab to click on New asset<\/strong> again.<\/p>\n
\"screenshot\"<\/a><\/p>\n
\n
As before, I clicked into DataStage component<\/strong> and then Schema library<\/strong>.<\/p>\n
\"screenshot\"<\/a>
\n\"screenshot\"<\/a><\/p>\n
\n
I named this schema library to show I’d use it to store the schema of output messages.<\/p>\n
\"screenshot\"<\/a><\/p>\n
\n
Then clicked on Add resource<\/strong> to upload my JSON file.<\/p>\n
\"screenshot\"<\/a><\/p>\n
As before, I didn’t upload a schema, I uploaded a file with a sample message payload and let DataStage generate a schema for me.<\/p>\n
I created a file called combined.json<\/code> and came up with this idea for a simple combined payload.<\/p>\n
{\n    \"timestamp\": 1668805140,\n    \"datetime\": \"2022-11-18 15:59:00\",\n    \"apple\": {\n        \"open\": 147.4,\n        \"high\": 147.58,\n        \"low\": 147.4,\n        \"close\": 147.57,\n        \"volume\": 56044\n    },\n    \"microsoft\": {\n        \"open\": 147.4,\n        \"high\": 147.58,\n        \"low\": 147.4,\n        \"close\": 147.57,\n        \"volume\": 56044\n    }\n}\n<\/pre>\n
(The numbers aren’t accurate – but they didn’t need to be. I just wanted a sample of the shape of the data that could be used to generate a schema).<\/p>\n
\n
DataStage used my combined.json<\/code> file to generate a combined.json.xsd<\/strong> schema.<\/p>\n
\"screenshot\"<\/a><\/p>\n
\n
Preparation done – now I was ready to create my flow.<\/p>\n
Time to click the New asset<\/strong> button for the last time.<\/p>\n
\"screenshot\"<\/a><\/p>\n
\n
This time I chose DataStage<\/strong>.<\/p>\n
\"screenshot\"<\/a><\/p>\n
\n
I gave my flow the name share-prices-combiner<\/strong>. Still feeling unimaginative.<\/p>\n
\"screenshot\"<\/a><\/p>\n
\n
Here was my shiny new canvas, ready to start setting up a flow.<\/p>\n
\"screenshot\"<\/a><\/p>\n
\n
Here is my finished flow.<\/p>\n
I’ll go through the critical settings I used for each node below. But first, at a high level, the flow:<\/p>\n