Icon

Kafka_​integration

Knime Integration with Kafka

This workflow represents a worfklow that receives information from kafka, perform operation on it and publish it back to kafka.

This workflow consumes transactions in JSON format from kafka. For the customer ID field in a record, the workflow fetches extra customer informaton from an already existing table and appends the information to the processing record. Also, the same is done for each product number in the record. Once, complete information is available in a single table, the workflow aggregates the data to calculate all the products bought by a single customer and the total amount spent by him. This aggregated information is passed back to kafka on topic "customer_transactions".

To run this worfklow :
1. Start Kafka cluster on your local system
2. Create a topic "transactions" and publish JSON records of the following form to it:
{"OrderNumber" : 23893756,"Date" : "8-28-2015","CustomerID" : "69-695-442-229","ProductNr" : "I-163-2017"}.
3. Once that is done, edit kafka consumer node to edit the stop criteria and change the time stamp value.
4. Now, run the workflow
5. Once the workflow is completed, run a kafka consumer on your local system and consume from the topic "customer_transactions". The processed records with aggregated information should be available on the new topic.

Consume Transaction records from kafka Append Customer and Product Information to each record Perform necessary aggregation and post the result back to kafka Connect to Kafka ClusterCustomer InformationProduct InformationConsume records from"transactions" topicParse JSON recordsFilter RelevantColumnsFor each customer,collect all the products boughtand total price Publish records back to kafka Kafka Connector Table Reader SQLite Connector DB Table Selector DB Reader Kafka Consumer Java Snippet Column Filter Column Filter GroupBy Table to JSON Kafka Producer Kafka Connector Add customerinformation Add ProductInformation Consume Transaction records from kafka Append Customer and Product Information to each record Perform necessary aggregation and post the result back to kafka Connect to Kafka ClusterCustomer InformationProduct InformationConsume records from"transactions" topicParse JSON recordsFilter RelevantColumnsFor each customer,collect all the products boughtand total price Publish records back to kafkaKafka Connector Table Reader SQLite Connector DB Table Selector DB Reader Kafka Consumer Java Snippet Column Filter Column Filter GroupBy Table to JSON Kafka Producer Kafka Connector Add customerinformation Add ProductInformation

Nodes

Extensions

Links

Download

To use this workflow in KNIME, download it from the below URL and open it in KNIME:

Download Workflow
Created by: knoldus
Created at: 2020-07-24
On NodePit since: 2020-08-06
Last update: 2024-12-17
Created with KNIME version: v4.2.0
Tags: Kafkakafka extension knimeETLpub-sub

Deploy, schedule, execute, and monitor your KNIME workflows locally, in the cloud or on-premises – with our brand new NodePit Runner.

Try NodePit Runner!