Hands-On with Flink — Part 6: Calling LLMs from Flink

Previous parts:

Part 1: Filtering data from Kafka

Part 2: Running on Kubernetes with the Operator

Part 3: SQL instead of Java

Part 4: SQL and Avro

Part 5: Managing State

Integrating ML Models into Data Pipelines

Let’s leave aside boring SELECTs and JOINs and do something more exciting.

In the latest versions of Apache Flink, machine learning models become first-class citizens and can be invoked directly from SQL pipelines. No custom Java code, no user-defined functions, and no external inference services. Everything can be done using pure Flink SQL.

In this part of the Hands-On with Flink series, we will build a simple system that summarizes product reviews using a Large Language Model (LLM).

The idea is illustrated in the diagram below.

Product reviews arrive as events in the product_reviews Kafka topic. Our first Flink job continuously aggregates reviews for each product and produces a compact context event into the product_review_context topic. These context events contain information such as the number of reviews, average rating, and the collected review texts.

A second Flink job consumes the aggregated context events and uses an LLM to generate a concise summary for each product. The generated summaries are then written to the product_review_summary topic.

This architecture demonstrates an important pattern when integrating AI with streaming systems: Flink is responsible for collecting, filtering, aggregating, and preparing the context, while the LLM focuses on interpreting that context and generating insights.

As usual, everything will run locally on our machine. And yes, that includes the LLM.

The source code for this part is here.

New Flink AI Constructs

Before diving into the installation steps, let’s first look at the Flink SQL features that we will use in this example.

The first part of the pipeline is relatively straightforward. We will read product reviews from Kafka and aggregate them into a compact product context. This context will later be sent to the LLM.

Reading raw reviews

First, we define a source table over the product_reviews Kafka topic:

CREATE TABLE product_reviews (
  review_id STRING,
  product_id STRING,
  product_name STRING,
  rating INT,
  review_text STRING,
  proc_time AS PROCTIME()
) WITH (
  'connector' = 'kafka',
  'topic' = 'product_reviews',
  'properties.bootstrap.servers' = 'host.docker.internal:9092',
  'properties.group.id' = 'product-reviews-source',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json'
);

This table represents raw review events arriving into Kafka. Most fields are self-explanatory: review ID, product ID, product name, rating, and review text.

The only new field is:

proc_time AS PROCTIME()

This is a computed column. It tells Flink to attach the current processing time to each incoming record. We will use this timestamp to group reviews into short time intervals.

Creating product context

Next, we define the output table for aggregated product context:

CREATE TABLE product_review_context (
  product_id STRING,
  product_name STRING,
  window_start TIMESTAMP(3),
  window_end TIMESTAMP(3),
  review_count BIGINT,
  average_rating DOUBLE,
  negative_reviews BIGINT,
  positive_reviews BIGINT,
  reviews STRING
) WITH (
  'connector' = 'kafka',
  'topic' = 'product_review_context',
  'properties.bootstrap.servers' = 'host.docker.internal:9092',
  'format' = 'json'
);

This topic will not contain individual reviews. Instead, it will contain aggregated context for each product.

For each product and time interval, Flink will produce information such as:

• how many reviews were received
• the average rating
• the number of negative reviews
• the number of positive reviews
• the review texts concatenated into one field

The window_start and window_end fields represent the time boundaries of the aggregation interval. We have not covered Flink windows in detail yet, so for now it is enough to understand them as the beginning and end of the time bucket in which reviews are collected.

Aggregating reviews

Finally, we define the streaming aggregation:

INSERT INTO product_review_context
SELECT
  product_id,
  MAX(product_name) AS product_name,
  window_start,
  window_end,
  COUNT(*) AS review_count,
  AVG(rating) AS average_rating,
  SUM(CASE WHEN rating <= 2 THEN 1 ELSE 0 END) AS negative_reviews,
  SUM(CASE WHEN rating >= 4 THEN 1 ELSE 0 END) AS positive_reviews,
  LISTAGG(
    CONCAT(
      'Rating: ',
      CAST(rating AS STRING),
      '. Review: ',
      review_text
    ),
    ' | '
  ) AS reviews
FROM TABLE(
  TUMBLE(
    TABLE product_reviews,
    DESCRIPTOR(proc_time),
    INTERVAL '1' MINUTE
  )
)
GROUP BY
  product_id,
  window_start,
  window_end;

This query continuously reads incoming reviews and groups them by product and by one-minute processing-time windows.

For each product in each one-minute interval, it calculates simple statistics and also creates a textual review context using LISTAGG.

This is important for the LLM part. We do not want to send every single review to the model separately. Instead, Flink first prepares a compact context event, and only this context event will be passed to the LLM.

The output event will look conceptually like this:

{
  "product_id": "P200",
  "product_name": "NoiseCancel Headphones",
  "review_count": 5,
  "average_rating": 4.0,
  "negative_reviews": 0,
  "positive_reviews": 4,
  "reviews": "Rating: 5. Review: Excellent sound quality... | Rating: 4. Review: Great noise cancellation..."
}

Creating a model

And now comes the interesting part.

We will define a model backed by Llama 3.1, running locally in Ollama. For this, Flink provides a new SQL construct:

CREATE MODEL

The model will take the aggregated product review context as input and return a generated summary as output.

In our case, the input field is:

reviews STRING

and the output field is:

content STRING

The model behavior is controlled by the system prompt:

You are a product analyst. Analyze the reviews and return a valid JSON only with fields: sentiment, key_strengths, key_complaints, summary, recommended_action.

Here is the complete model definition:

CREATE MODEL product_review_summarizer
INPUT (reviews STRING)
OUTPUT (content STRING)
WITH (
  'provider' = 'openai',
  'endpoint' = 'http://ollama:11434/v1/chat/completions',
  'api-key' = 'ollama',
  'model' = 'llama3.1',
  'system-prompt' = 'You are a product analyst. Analyze the reviews and return a valid JSON only with fields: sentiment, key_strengths, key_complaints, summary, recommended_action.'
);

Although the provider is configured as openai, we are not calling OpenAI here. Ollama exposes an OpenAI-compatible API, so Flink can use the same provider interface to call our locally running model.

The endpoint:

http://ollama:11434/v1/chat/completions

points to the Ollama service inside the Kubernetes cluster.

Creating the summary sink

The final piece of our pipeline is a sink table for the summaries generated by the LLM.

The table is backed by the product_review_summary Kafka topic. Each record will contain the original product statistics together with the summary returned by the model.

CREATE TABLE product_review_summary (
  product_id STRING,
  product_name STRING,
  review_count BIGINT,
  average_rating DOUBLE,
  negative_reviews BIGINT,
  positive_reviews BIGINT,
  llm_summary STRING
) WITH (
  'connector' = 'kafka',
  'topic' = 'product_review_summary',
  'properties.bootstrap.servers' = 'host.docker.internal:9092',
  'format' = 'json'
);

At this point, we have all the building blocks in place:

1. A source table reading raw reviews from Kafka.
2. An aggregation job producing product context.
3. A model definition backed by Llama 3.1 running in Ollama.
4. A sink table for storing generated summaries.

Invoking the ML Model from Flink

The only missing piece is connecting the product context with the model. This is where the second new Flink AI construct comes into play:

ML_PREDICT

Just as CREATE MODEL defines a model, ML_PREDICT invokes it directly from a SQL pipeline.

In this step, we read aggregated product context events from Kafka, call the LLM for each context event, and write the generated summary back to Kafka.

INSERT INTO product_review_summary
SELECT
  product_id,
  product_name,
  review_count,
  average_rating,
  negative_reviews,
  positive_reviews,
  content AS llm_summary
FROM ML_PREDICT(
  TABLE product_review_context,
  MODEL product_review_summarizer,
  DESCRIPTOR(reviews)
);

The important part is this:

ML_PREDICT(
  TABLE product_review_context,
  MODEL product_review_summarizer,
  DESCRIPTOR(reviews)
)

Here, product_review_context is the input table, product_review_summarizer is the model defined earlier, and reviews is the field passed to the model as input.

For every product context event, Flink calls the LLM and receives a generated response in the content field. We then rename this field to llm_summary and write the result into the product_review_summary Kafka topic.

Now that we understand the architecture and the new Flink AI constructs, it’s time to put everything into practice.

Fair warning: there is quite a bit of setup involved.

Reviews → Kafka → Flink → LLM → Kafka

As we work through the deployment, don’t lose sight of the main idea behind the example:

Flink creates the context. The LLM explains it.

This separation of responsibilities is what makes the combination so powerful. Flink excels at collecting, filtering, aggregating, and enriching streaming data, while the LLM focuses on interpreting the prepared context and generating insights.

Upgrading the Flink Kubernetes Operator

In this article, we are using some of the newest Flink SQL features. Therefore, we also need a recent version of the Flink Kubernetes Operator.

Assuming that the operator is not installed yet, we first add the Helm repository for version 1.15.0:

helm repo add flink-operator-repo \

  https://downloads.apache.org/flink/flink-kubernetes-operator-1.15.0/

helm repo update

Next, we install the required Custom Resource Definitions (CRDs).

In my local setup, the Helm chart did not install all CRDs that were needed by the operator, so I added the missing ones manually:

kubectl apply -f https://raw.githubusercontent.com/apache/flink-kubernetes-operator/main/helm/flink-kubernetes-operator/crds/flinkdeployments.flink.apache.org-v1.yml

kubectl apply -f https://raw.githubusercontent.com/apache/flink-kubernetes-operator/main/helm/flink-kubernetes-operator/crds/flinkstatesnapshots.flink.apache.org-v1.yml

kubectl apply -f https://raw.githubusercontent.com/apache/flink-kubernetes-operator/main/helm/flink-kubernetes-operator/crds/flinkbluegreendeployments.flink.apache.org-v1.yml

Finally, we install the operator itself:

helm install flink-kubernetes-operator \
  flink-operator-repo/flink-kubernetes-operator \
  --namespace flink \
  --create-namespace \
  --set webhook.create=false

I disable the webhook because this is a local development environment and I want to keep the setup as simple as possible.

Preparing Kafka Topics

As usual, we will run Kafka in Docker. For this exercise, we only need a Kafka broker. Since we will use JSON for serialization, there is no need to deploy Schema Registry.

Our docker-compose.yml contains two containers:

• Kafka — the message broker
• AKHQ — a web-based UI for inspecting topics, messages, and consumer groups

Let’s start the environment:

docker compose up -d

Next, create the Kafka topics that will be used throughout the example:

• product_reviews – incoming product reviews
• product_review_context – aggregated review context produced by Flink
• product_review_summary – summaries generated by the LLM

The topics can be created using AKHQ tool or from the scripts inside the Kafka container. We will use these topics throughout the rest of the article as data flows from raw reviews, through Flink aggregations, and finally into LLM-generated summaries.

Preparing the LLM

For this example, we will use Ollama to run the LLM locally. Instead of calling an external cloud API, we will deploy Ollama directly into our Kubernetes environment.

The ollama.yml descriptor contains:

• one Kubernetes Deployment using the Ollama image
• one Kubernetes Service exposing Ollama inside the cluster

Let’s install it into the flink namespace:

kubectl apply -f ollama.yml -n flink

After the pod is running, we also need to pull the actual model. Ollama provides the runtime, but the model itself has to be downloaded separately.

For this example, we will use Llama 3.1:

kubectl exec -it deployment/ollama -n flink -- ollama pull llama3.1

We can test ollama directly by using port-forwarding

kubectl port-forward svc/ollama 11434:11434 -n flink

In another terminal:

curl http://localhost:11434

Expected response:

Ollama is running

Later, after we deploy the Flink SQL session, we can verify connectivity from Flink to Ollama.

Building a Custom Flink Image

To use Flink’s new AI capabilities, we need a custom Flink image containing several additional libraries:

• Kafka client library
• Kafka connector for Flink
• OpenAI model connector, which enables the CREATE MODEL and ML_PREDICT SQL statements

The Dockerfile looks as follows:

FROM flink:2.1.2-java17

USER root

ARG FLINK_VERSION=2.1.2
ARG KAFKA_CONNECTOR_VERSION=4.0.1-2.0
ARG KAFKA_CLIENTS_VERSION=3.9.0

# Kafka connector
ADD https://repo1.maven.org/maven2/org/apache/flink/flink-connector-kafka/${KAFKA_CONNECTOR_VERSION}/flink-connector-kafka-${KAFKA_CONNECTOR_VERSION}.jar /opt/flink/lib/

ADD https://repo1.maven.org/maven2/org/apache/kafka/kafka-clients/${KAFKA_CLIENTS_VERSION}/kafka-clients-${KAFKA_CLIENTS_VERSION}.jar /opt/flink/lib/

# Flink AI / OpenAI model connector
ADD https://repo1.maven.org/maven2/org/apache/flink/flink-model-openai/${FLINK_VERSION}/flink-model-openai-${FLINK_VERSION}.jar /opt/flink/lib/

# Fix ownership
RUN chown flink:flink /opt/flink/lib/*.jar

USER flink

The most important dependency is flink-model-openai. Despite its name, it is not limited to OpenAI models. It works with any OpenAI-compatible API, including Ollama, which is exactly what we will use later in this article.

Once the Dockerfile is ready, build the image:

docker build -t localhost:5001/flink-ai-demo:2.1.2 .

And push it to the local registry:

docker push localhost:5001/flink-ai-demo:2.1.2

The image is now ready to be used by our Flink SQL session cluster.

Deploying the Flink Cluster

Now we can deploy the Flink cluster into our Kubernetes environment.

We already know the basic deployment process from the previous parts, but this time there are two important changes in the deployment descriptors:

1. We use our custom image, localhost:5001/flink-ai-demo:2.1.2, for both the Flink SQL session cluster and the SQL Gateway.
2. In the FlinkDeployment, we set the Flink version to:

flinkVersion: v2_1

This is important because Flink 2.1 uses the newer configuration format and also provides the SQL features we need for model inference.

After updating the descriptors, we deploy the Flink SQL session cluster:

kubectl apply -f flink-sql-deployment.yaml -n flink

Then we deploy the SQL Gateway:

kubectl apply -f sql-gateway.yaml -n flink

Once both components are running, we will be able to submit Flink SQL statements through the SQL Gateway API.

Finally, let’s verify that everything is running correctly. Check the namespace:

kubectl get pods -n flink

You should see something similar to:

NAME                                         READY   STATUS    RESTARTS   AGE
flink-kubernetes-operator-7b7c9bbfb8-kdnkj   1/1     Running   0          85s
flink-sql-gateway-7969db4f45-8jq46           1/1     Running   0          3s
flink-sql-session-84c796f566-798qt           1/1     Running   0          82s
ollama-947448f9f-8nwgg                       1/1     Running   0          21h

If all pods are in the Running state, we are ready to start submitting SQL statements and building our AI-powered streaming pipeline.

Deploying the Data Pipeline

Now let’s deploy the Flink SQL queries that we discussed earlier.

First, start port-forwarding for the SQL Gateway so that we can submit SQL statements over HTTP:

kubectl port-forward deployment/flink-sql-gateway 8083:8083 -n flink

Open another terminal and create a SQL Gateway session:

curl -X POST http://localhost:8083/v1/sessions \
  -H 'Content-Type: application/json' \
  -d '{}'

You should receive a response similar to this:

{

  "sessionHandle": "4e92613e-9f4a-4ac7-82bf-34a35c4c73ad"

}

Use your own sessionHandle in the following commands.

Create the source table:

curl -X POST http://localhost:8083/v1/sessions/4e92613e-9f4a-4ac7-82bf-34a35c4c73ad/statements \
  -H "Content-Type: application/json" \
  -d '{
    "statement": "CREATE TABLE product_reviews (\n  review_id STRING,\n  product_id STRING,\n  product_name STRING,\n  rating INT,\n  review_text STRING,\n  proc_time AS PROCTIME()\n) WITH (\n  '\''connector'\'' = '\''kafka'\'',\n  '\''topic'\'' = '\''product_reviews'\'',\n  '\''properties.bootstrap.servers'\'' = '\''host.docker.internal:9092'\'',\n  '\''properties.group.id'\'' = '\''product-reviews-source'\'',\n  '\''scan.startup.mode'\'' = '\''earliest-offset'\'',\n  '\''format'\'' = '\''json'\''\n);"
  }'

Create the product context table:

curl -X POST http://localhost:8083/v1/sessions/4e92613e-9f4a-4ac7-82bf-34a35c4c73ad/statements \
  -H "Content-Type: application/json" \
  -d '{
    "statement": "CREATE TABLE product_review_context (\n  product_id STRING,\n  product_name STRING,\n  window_start TIMESTAMP(3),\n  window_end TIMESTAMP(3),\n  review_count BIGINT,\n  average_rating DOUBLE,\n  negative_reviews BIGINT,\n  positive_reviews BIGINT,\n  reviews STRING\n) WITH (\n  '\''connector'\'' = '\''kafka'\'',\n  '\''topic'\'' = '\''product_review_context'\'',\n  '\''properties.bootstrap.servers'\'' = '\''host.docker.internal:9092'\'',\n  '\''properties.group.id'\'' = '\''product-review-context-llm'\'',\n  '\''scan.startup.mode'\'' = '\''earliest-offset'\'',\n  '\''format'\'' = '\''json'\''\n);"
  }'

This table is used twice: first as a sink for the aggregation job and later as a source for ML_PREDICT. Because it will be consumed by the second job, we also define a Kafka consumer group using properties.group.id.

Create the aggregation job:

curl -X POST http://localhost:8083/v1/sessions/4e92613e-9f4a-4ac7-82bf-34a35c4c73ad/statements \
  -H "Content-Type: application/json" \
  -d '{
    "statement": "INSERT INTO product_review_context SELECT product_id, MAX(product_name) AS product_name, window_start, window_end, COUNT(*) AS review_count, AVG(rating) AS average_rating, SUM(CASE WHEN rating <= 2 THEN 1 ELSE 0 END) AS negative_reviews, SUM(CASE WHEN rating >= 4 THEN 1 ELSE 0 END) AS positive_reviews, LISTAGG(CONCAT('\''Rating: '\'', CAST(rating AS STRING), '\''. Review: '\'', review_text), '\'' | '\'') AS reviews FROM TABLE(TUMBLE(TABLE product_reviews, DESCRIPTOR(proc_time), INTERVAL '\''1'\'' MINUTE)) GROUP BY product_id, window_start, window_end;"
  }'

This statement starts a long-running Flink job. It continuously reads raw reviews, aggregates them into one-minute product contexts, and writes the result to the product_review_context topic.

Create the model:

curl -X POST http://localhost:8083/v1/sessions/4e92613e-9f4a-4ac7-82bf-34a35c4c73ad/statements \
  -H "Content-Type: application/json" \
  -d '{
    "statement": "CREATE MODEL product_review_summarizer INPUT (reviews STRING) OUTPUT (content STRING) WITH ( '\''provider'\'' = '\''openai'\'', '\''endpoint'\'' = '\''http://ollama:11434/v1/chat/completions'\'', '\''api-key'\'' = '\''ollama'\'', '\''model'\'' = '\''llama3.1'\'', '\''system-prompt'\'' = '\''You are a product analyst. Analyze the reviews and return a valid JSON only with fields: sentiment, key_strengths, key_complaints, summary, recommended_action.'\'' );"
  }'

Create the summary sink:

curl -X POST http://localhost:8083/v1/sessions/4e92613e-9f4a-4ac7-82bf-34a35c4c73ad/statements \
  -H "Content-Type: application/json" \
  -d '{
    "statement": "CREATE TABLE product_review_summary (\n  product_id STRING,\n  product_name STRING,\n  review_count BIGINT,\n  average_rating DOUBLE,\n  negative_reviews BIGINT,\n  positive_reviews BIGINT,\n  llm_summary STRING\n) WITH (\n  '\''connector'\'' = '\''kafka'\'',\n  '\''topic'\'' = '\''product_review_summary'\'',\n  '\''properties.bootstrap.servers'\'' = '\''host.docker.internal:9092'\'',\n  '\''format'\'' = '\''json'\''\n);"
  }'

Starting the LLM inference job:

curl -X POST http://localhost:8083/v1/sessions/4e92613e-9f4a-4ac7-82bf-34a35c4c73ad/statements \
  -H "Content-Type: application/json" \
  -d '{
    "statement": "INSERT INTO product_review_summary SELECT product_id, product_name, review_count, average_rating, negative_reviews, positive_reviews, content AS llm_summary FROM ML_PREDICT(TABLE product_review_context, MODEL product_review_summarizer, DESCRIPTOR(reviews));"
  }'

This starts the second long-running Flink job. It reads product context events, invokes the LLM using ML_PREDICT, and writes the generated summaries into the product_review_summary topic.

Generating the Input Data

Now let’s put the pipeline to work.

The repository contains a helper script, reviews.sh, which generates sample product reviews and sends them to the product_reviews topic:

./reviews.sh

The reviews will be consumed by the first Flink job, aggregated into product context, and then analyzed by the LLM.

Since the aggregation uses a one-minute window, give the pipeline a little time to process the data. After a minute or so, inspect the product_review_summary topic using AKHQ.

You should see records similar to the following:

{
  "product_id": "P200",
  "product_name": "NoiseCancel Headphones",
  "review_count": 5,
  "average_rating": 4.0,
  "negative_reviews": 0,
  "positive_reviews": 4,
  "llm_summary": "Here is a valid JSON structure with sentiment, key_strengths, key_complaints, summary, and recommended_action:\n\n{\n  \"sentiment\": \"Positive\",\n  \"key_strengths\": [\"Excellent sound quality\", \"Very comfortable\", \"Great noise cancellation\", \"Battery lasts for days\"],\n  \"key_complaints\": [\"Case is bulky\", \"Bluetooth pairing was sometimes unstable\", \"Touch controls are too sensitive\"],\n  \"summary\": \"Overall, this product has received mostly positive reviews, with users praising its comfort and sound quality. However, some have experienced issues with the case and Bluetooth pairing.\",\n  \"recommended_action\": \"Improve case design for a more compact fit\"\n}"
}

This event contains both the aggregated review statistics and the analysis generated by the LLM.

The most interesting part is the JSON generated by the model itself:

{
  "sentiment": "Positive",
  "key_strengths": [
    "Excellent sound quality",
    "Very comfortable",
    "Great noise cancellation",
    "Battery lasts for days"
  ],
  "key_complaints": [
    "Case is bulky",
    "Bluetooth pairing was sometimes unstable",
    "Touch controls are too sensitive"
  ],
  "summary": "Overall, this product has received mostly positive reviews, 
              with users praising its comfort and sound quality. However, 
              some have experienced issues with the case and Bluetooth pairing.",
  "recommended_action": "Improve case design for a more compact fit"
}

This simple example demonstrates the core idea behind combining Flink and LLMs:

Flink creates the context. The LLM explains it.

In this article we combined Kafka, Flink, and a locally running LLM into a complete streaming AI pipeline. While the example focused on product reviews, the same pattern can be applied to customer interactions, incident management, fraud detection, log analysis, and many other domains. The key idea remains the same: Flink prepares the context, and the LLM interprets it.

Happy exploring!