Real-Time Text Search and Analytics with confluent-kafka and nlp4j-local-search

Apache Kafka is often used to process continuously arriving events such as application logs, customer activity, sensor data, and messages.

But what if the events contain natural-language text?

In this article, we combine:

• confluent-kafka for consuming Kafka events
• nlp4j-local-search for local full-text search and text analytics

The basic idea is simple:

Kafka Topic
    |
    v
confluent-kafka Consumer
    |
    v
Text message
    |
    v
nlp4j-local-search
    |
    +--> Full-text search
    |
    +--> Text analytics

This makes it possible to search and analyze text shortly after it arrives from Kafka.

Example Use Case: Customer Feedback

Suppose customer feedback messages are continuously sent to a Kafka topic called:

customer-feedback

Each Kafka message contains JSON like this:

{
  "customer_id": "C001",
  "text": "The brake makes a loud noise when stopping."
}

Other messages may look like:

{
  "customer_id": "C002",
  "text": "The battery does not charge completely."
}

{
  "customer_id": "C003",
  "text": "I hear a strange noise from the brake."
}

A Kafka consumer can receive these messages and immediately add the text to a local search index.

Installation

Install both Python packages:

pip install confluent-kafka
pip install nlp4j-local-search==0.5.0

Then import them:

from confluent_kafka import Consumer
from nlp4j_local_search import SearchEngine

Producing Sample Messages

First, here is a small producer for testing.

import json
from confluent_kafka import Producer

producer = Producer({
    "bootstrap.servers": "localhost:9092"
})

messages = [
    {
        "customer_id": "C001",
        "text": "The brake makes a loud noise when stopping."
    },
    {
        "customer_id": "C002",
        "text": "The battery does not charge completely."
    },
    {
        "customer_id": "C003",
        "text": "I hear a strange noise from the brake."
    },
    {
        "customer_id": "C004",
        "text": "The engine makes a strange sound."
    },
    {
        "customer_id": "C005",
        "text": "The brake pedal feels soft."
    },
]

for message in messages:
    producer.produce(
        "customer-feedback",
        value=json.dumps(message).encode("utf-8")
    )

producer.flush()

print("Messages sent.")

This is just a normal Kafka producer.

There is nothing specific to NLP4J on the producer side.

Consuming Messages and Creating a Search Index

Now let's create a Kafka consumer.

Whenever a message arrives, we add its text to SearchEngine.

import json

from confluent_kafka import Consumer, KafkaException
from nlp4j_local_search import SearchEngine

consumer = Consumer({
    "bootstrap.servers": "localhost:9092",
    "group.id": "nlp4j-text-analysis",
    "auto.offset.reset": "earliest",
    "enable.auto.commit": False,
})

consumer.subscribe(["customer-feedback"])

with SearchEngine("en") as engine:

    try:

        while True:

            msg = consumer.poll(1.0)

            if msg is None:
                continue

            if msg.error():
                raise KafkaException(msg.error())

            event = json.loads(
                msg.value().decode("utf-8")
            )

            text = event["text"]

            # Kafka topic / partition / offset can be used
            # as a unique document identifier.
            document_id = (
                f"{msg.topic()}-"
                f"{msg.partition()}-"
                f"{msg.offset()}"
            )

            engine.add(document_id, text)
            engine.commit()

            print("Received:", text)

            consumer.commit(
                message=msg,
                asynchronous=False
            )

    finally:
        consumer.close()

The important part is only:

engine.add(document_id, text)
engine.commit()

Each Kafka message becomes a searchable document.

Searching Incoming Messages

We can search the documents already received by the consumer.

For example:

results = engine.search(
    "brake",
    limit=10
)

for result in results:
    print(result.id, result.body)

The result may contain messages such as:

The brake makes a loud noise when stopping.
I hear a strange noise from the brake.
The brake pedal feels soft.

The interesting point is that these documents originally arrived as Kafka events.

They did not need to be exported to another search server before they could be searched.

The processing flow is:

Kafka
  |
  | customer-feedback
  v
Python Consumer
  |
  | engine.add(...)
  v
LocalSearch
  |
  | engine.search("brake")
  v
Search Results

Running Text Analytics on the Stream

nlp4j-local-search 0.5.0 can also analyze the indexed text with view().

For example, suppose we want to know:

What nouns are characteristic of feedback messages mentioning "brake"?

We can run:

result = engine.view(
    query_field="word.noun",
    query_value="brake",
    field="word.noun",
    size=100,
)

print(
    "Target documents:",
    result.count,
    "/",
    result.total_count
)

for bucket in result.buckets:
    print(
        bucket.key,
        bucket.count,
        bucket.all_count,
        f"{bucket.relative_rate:.4f}"
    )

view() compares the selected documents with the entire indexed document collection.

Conceptually:

relative_rate
    =
term rate in target documents
    /
term rate in all documents

If a word appears much more frequently in documents about brake than in the whole stream, its relative_rate becomes higher.

This allows us to move from:

Search for messages containing "brake"

to:

Find words characteristic of messages about "brake"

using the same local index.

Searching Every Few Messages

For demonstration purposes, calling:

engine.commit()

for every Kafka message is easy to understand.

For higher-throughput processing, however, it is usually better to process messages in batches.

For example:

message_count = 0

while True:

    msg = consumer.poll(1.0)

    if msg is None:
        continue

    if msg.error():
        raise KafkaException(msg.error())

    event = json.loads(
        msg.value().decode("utf-8")
    )

    document_id = (
        f"{msg.topic()}-"
        f"{msg.partition()}-"
        f"{msg.offset()}"
    )

    engine.add(
        document_id,
        event["text"]
    )

    message_count += 1

    if message_count % 100 == 0:

        engine.commit()

        results = engine.search(
            "brake",
            limit=10
        )

        print(
            "Indexed:",
            message_count,
            "messages"
        )

The appropriate batch size depends on the application.

The important idea is that Kafka event ingestion and Lucene indexing do not need to be one-to-one operations.

Using Confluent Cloud

The NLP4J part does not change when using Confluent Cloud.

Only the Kafka consumer configuration changes.

For example:

consumer = Consumer({
    "bootstrap.servers": "<BOOTSTRAP_SERVER>",
    "security.protocol": "SASL_SSL",
    "sasl.mechanism": "PLAIN",
    "sasl.username": "<API_KEY>",
    "sasl.password": "<API_SECRET>",
    "group.id": "nlp4j-text-analysis",
    "auto.offset.reset": "earliest",
})

The rest of the application can remain:

Consumer
   |
   v
JSON event
   |
   v
SearchEngine.add()
   |
   +--> search()
   |
   +--> view()

Do not put Confluent Cloud credentials directly into source code in a real application. Environment variables or an appropriate secrets-management mechanism should be used instead.

Kafka + NLP4J

The responsibilities of the two libraries are different.

confluent-kafka handles the event stream:

Kafka brokers
Topics
Partitions
Offsets
Consumer groups

nlp4j-local-search handles the text:

Text analysis
Lucene indexing
Full-text search
Aggregation
Characteristic-term analysis

Combining them produces a small architecture like this:

                         +-------------------+
                         |   Kafka Producer  |
                         +---------+---------+
                                   |
                                   v
                         +-------------------+
                         |    Kafka Topic    |
                         | customer-feedback |
                         +---------+---------+
                                   |
                                   v
                         +-------------------+
                         | confluent-kafka   |
                         |     Consumer      |
                         +---------+---------+
                                   |
                                   v
                         +-------------------+
                         | nlp4j-local-search|
                         +---------+---------+
                                   |
                    +--------------+--------------+
                    |                             |
                    v                             v
               search()                       view()
                    |                             |
                    v                             v
              Find messages              Analyze terms

Why This Combination Is Interesting

A common architecture would send Kafka events to an external search system.

That approach is appropriate for large distributed systems.

But not every application requires a separate Elasticsearch, OpenSearch, or Solr cluster.

For example:

• local analytics applications
• edge processing
• experimental pipelines
• development tools
• small internal services
• NLP prototypes

may only need a local searchable index.

In such cases, the application can consume events directly:

msg = consumer.poll(...)

and index them directly:

engine.add(...)

without introducing another search server.

A Small Real-Time NLP Pipeline

The entire concept can be summarized as:

msg = consumer.poll(1.0)

event = json.loads(
    msg.value().decode("utf-8")
)

engine.add(
    document_id,
    event["text"]
)

engine.commit()

results = engine.search(
    "brake"
)

Kafka provides the stream.

NLP4J provides text analysis and search.

The combination makes it possible to build a simple real-time text search and analytics pipeline entirely from Python.

Summary

Using:

confluent-kafka
+
nlp4j-local-search

we can build a pipeline that:

1. receives text events from Kafka
2. performs NLP processing
3. indexes the text with Lucene
4. searches recently received messages
5. analyzes characteristic terms with view()

The basic integration requires no special adapter.

A Kafka message simply becomes input to:

engine.add(...)

This means nlp4j-local-search can be used not only for static datasets, but also as a text-analysis component in Kafka-based event-processing applications.

For small real-time NLP applications, experiments, and local analytics, this can provide a simple alternative to introducing a separate search server.