Ctrlk
For the complete documentation index, see llms.txt. This page is also available as Markdown.

Use Cases

Use Cases

Complete hands-on workshops for learning Kafka-based streaming data integration with Pentaho Data Integration (PDI) Kafka Enterprise Edition plugin.

Scenario: Basic Kafka Consumer - Real-time User Activity Stream

Kafka PlugIn

x

  1. Ensure the Kafka EE plugin is installed.

Kafka EE plugin

  1. Start Pentaho Data Integration.

Select a Use Case:

Scenario: Basic Kafka Consumer - Real-time User Activity Stream

Your company tracks user registrations across web and mobile platforms. User registration events are published to a Kafka topic in real-time.

Your task is to build a streaming pipeline that continuously reads these events, parses the JSON payload, transforms timestamps, and loads the data into a MySQL data warehouse - enabling real-time dashboards and analytics.

Kafka Consumer

The Kafka Consumer step pulls streaming data from Kafka through a transformation. Within the Kafka Consumer step you enter the path that will execute the transformation according to message batch size or duration in near real-time. The child transformation must start with the Get records from stream step.

Additionally, from the Kafka Consumer step, you can select a step in the child transformation to stream records back to the parent transformation. This allows records processed by a Kafka Consumer step in a parent transformation to be passed downstream to any other steps included within the same parent transformation.

Architecture Overview

This workshop uses PDI's parent/child transformation pattern for Kafka streaming:

How it works: The parent transformation's Kafka Consumer step reads messages in batches (every 5 seconds or 100 records, whichever comes first) and passes each batch to the child transformation for processing. The child transformation parses, transforms, and writes each batch to MySQL.

  1. Open the following transformation:

~/Workshop--Data-Integration/Labs/Module 7 - Use Cases/Streaming Data/Kafka/transformations/users-to-db-parent.ktr

  1. Double-click on the Kafka Consumer step to review the properties:

Setup Tab

Setup
Property
Description
Value

Transformation

Child transformation to process the records

${Internal.Entry.Current.Directory}/users-to-db-child.ktr

Setup

Connection

Direct: Specify Bootstrap servers.

Cluster: Specify a Hadoop cluster configuration.

localhost:9092

Topics

Kafka topics to consume from

pdi-users

Consumer Group

Each Kafka consumer step starts a single thread. When part of a consumer group, each consumer is assigned a subset of topic partitions.

pdi-warehouse-users

Batch tab

How batching works: Whichever threshold is reached first (duration or record count) triggers the batch to be sent to the child transformation. With pdi-users producing ~1 msg/sec, the 5-second duration will usually trigger first, sending ~5 records per batch.

Batch
Property
Description
Value

Duration (ms)

Time (in milliseconds) to collect records before executing the child transformation.

500

Number of records

Number of records to collect before executing the child transformation.

100

Maximum concurrent batches

Maximum number of batches to collect at the same time.

1

Message prefetch limit

Limit for incoming messages to queue for processing.

100000

Offset Management

Commit when record read: Commit offset when a record is read.

Commit when batch completed: Commit offsets after the batch is processed.

Commit when batch completed

Fields tab

Fields
Property
Description
Value

Input Name

Incoming fields received from Kafka streams. Default inputs include:

key: Determines message distribution to partitions. If no key is present, messages are randomly distributed.

message: The message value.

topic: Topic name.

partition: Partition number.

offset: Sequential ID that uniquely identifies the record within the partition.

timestamp: Time the message is received on the server.

Output Name

Output field name.

Type

Data Type

Results fields tab

Results fields
Property
Description
Value

Return fields from

Step name in the child transformation that returns fields to the parent transformation.

Options tab

Options
Property
Description
Value

auto.offset.reset

set the offset from when to process the records: latest, earliest

earliest

Get records from stream

This step returns records that were previously generated by another transformation in a job. The records are passed to this step using one of the streaming input steps (for example, Kinesis consumer, Kafka consumer, etc.).

This step produces one or more rows and cannot be placed within a stream. It must be the first step in a stream since it produces rows. If you want to add this data to an existing stream, you need to use a join step.

users-to-db-child.ktr

  1. Open the following transformation:

~/Workshop--Data-Integration/Labs/Module 7 - Use Cases/Streaming Data/Kafka/transformations/users-to-db-child.ktr

Get records from stream

Receive batched records from users-to-db-parent.ktr

  1. Double-click on the step and enter the following Fieldnames and Types:

Stream feilds

JSON Input

Use the JSON Input step to read data from JSON structures, files, or incoming fields.

The step uses a JSONPath expression to extract data and output rows. JSONPath expressions can use either dot notation or square bracket notation.

  1. Double-click on the JSON Input step to display the settings:

File tab

  1. Ensure: Source is from previous step - enabled.

  2. From the drop down select: message field.

Content tab

  1. Suppress errors

Content

Fields tab

  1. You will have to manually enter the path.

Enter path to retrieve fields

You can get an idea of the JSON object by viewing the messages in the Control Center.

The pdi-users topic receives user registration events at ~1 message/second from the datagen connector.

Sample message:

Field descriptions:

JSON Field
Type
Description

registertime

Long

Registration timestamp (epoch milliseconds)

userid

String

User identifier (e.g., User_1)

regionid

String

Region identifier (e.g., Region_9)

gender

String

Gender (MALE or FEMALE)

Select values

  1. Double-click on the step to diaply the properties.

  2. Click on the Metadata tab.

Define Metadata

This tab sets the data type and length metadata for each field. This is critical for MySQL - without explicit lengths, PDI maps String fields to TINYTEXT, which breaks MySQL indexes and causes errors like:

Fieldname
Type
Length

user_id

String

100

region_id

String

100

gender

String

20

register_time_epoch

Integer

15

kafka_topic

String

255

kafka_partition

Integer

9

kafka_offset

Integer

15

key

String

100

message

String

5000

timestamp

Integer

15

Why these lengths? They match the MySQL table column definitions: user_id VARCHAR(100), region_id VARCHAR(100), gender VARCHAR(20), kafka_topic VARCHAR(255). Setting the correct lengths ensures PDI generates VARCHAR instead of TINYTEXT.

Formula

The Formula step can calculate Formula Expressions within a data stream. It can be used to create simple calculations like [A]+[B] or more complex business logic with a lot of nested if / then logic.

Formula

  1. Double-click to configure the Formula step:

New field
Formula
Value type
Length
Precision
Replace

register_time_seconds

[register_time_epoch] / 1000

Integer

-1

-1

(blank)

Why Formula instead of Calculator? The Calculator step requires both operands to be existing stream fields — you cannot enter a literal constant like 1000 as Field B. The Formula step supports inline constants in expressions.

What this does: The datagen produces registertime as epoch milliseconds (e.g., 1493899960000). MySQL's TIMESTAMP column expects epoch seconds, so we divide by 1000 to get 1493899960.

Alternative using Calculator: Add an Add constants step before Calculator with a field divisor = 1000 (Integer). Then use Calculator with operation A / B where A = register_time_epoch and B = divisor.

Table Output

The Table Output step loads data into a database table. Table Output is equivalent to the SQL INSERT operator.

If you only need to update rows, use the Update step.

To perform both INSERT and UPDATE, use Insert/Update.

This step provides configuration options for a target table and performance-related options such as Commit size and Use batch update for inserts.

Create Database Connection in Spoon

  1. Open Spoon (PDI)

  2. Go to View panel (left side) → right-click Database connections → New

  3. Configure:

Setting
Value

Connection Name

warehouse_db

Connection Type

MySQL

Access

Native (JDBC)

Host Name

localhost

Database Name

kafka_warehouse

Port Number

3306

User Name

kafka_user

Password

kafka_password

  1. Click the Options tab and add these parameters:

Parameter
Value

useServerPrepStmts

false

rewriteBatchedStatements

true

cachePrepStmts

true

prepStmtCacheSize

250

useCompression

true

  1. Click Test — should show "Connection successful"

  2. Click OK to save

Main Settings

  1. Double-click on the Table output stepto configure:

Setting
Value
Notes

Connection

warehouse_db

The MySQL connection from Step 2

Target schema

(leave blank)

Important: Do NOT set this for MySQL

Target table

user_events

Commit size

1000

Truncate table

No

Ignore insert errors

No

Use batch updates

Yes

Specify database fields

Yes

Must be Yes to control field mapping

Critical: Leave Target schema blank. MySQL uses the database name from the connection, not a separate schema. Setting it to kafka_warehouse causes PDI to qualify the table as kafka_warehouse.user_events which can fail or cause unexpected behavior.

Database Fields

Click Specify database fields: Yes, then configure the field mapping:

Database Column
Stream Field

user_id

user_id

region_id

region_id

gender

gender

register_time

register_time_seconds

kafka_topic

kafka_topic

kafka_partition

kafka_partition

kafka_offset

kafka_offset

Do NOT map these columns - MySQL handles them automatically:

  • event_id - AUTO_INCREMENT primary key

  • ingestion_timestamp - DEFAULT CURRENT_TIMESTAMP

Tip: You can use Get Fields button to auto-populate, then remove event_id and ingestion_timestamp, and fix the register_time mapping (stream field should be register_time_seconds, not register_time).

SQL Button

When you click SQL in the Table output dialog, PDI may suggest ALTER TABLE statements. Click Close without executing - the table already has the correct schema from the Docker init script.

If PDI suggests:

This means the Meta-data tab in Select values doesn't have the string lengths set.

Prerequisites

Before using the transformation templates, ensure:

Kafka cluster is running: make start or docker compose up -d from the Kafka-Docker directory

Datagen connectors are deployed: make deploy-connectors or ./connectors/deploy-connectors.sh

MySQL database is running: make mysql-setup

Run the following commands:

  1. Starts all the required Kafka containers + connectors + MySQL.

User event data is now being streamed to the Brokers

  1. Start users-to-db-parent.ktr

The user events are being consumed and processed, writing the stream to the user_events table.

  1. In DBeaver display the data in the user_events table.

View data - DBeaver

Verify Data in MySQL

E-Commerce Purchases

Your e-commerce platform publishes purchase transactions to Kafka using Avro serialization with a schema managed by Confluent Schema Registry. Avro provides schema enforcement at the producer — incompatible schema changes are rejected before data enters the pipeline, unlike raw JSON where structural issues are only discovered at parse time.

Your task is to configure the PDI Kafka Consumer to deserialize Avro messages, extract nested JSON fields (the address object), and load purchase data into MySQL.

x

x

x

x

x

x

PreviousKafka
NextJenkins

Last updated

Was this helpful?