Confluent

data processing
ai agent
developer tool

Introduction

Help customers shift left by bringing processing and cleaning closer to the data source, eliminating wasteful data, manual break-fix, and high costs.

payments

Id
uuid
tripID
varchar
method
enum
amount
float
createdAt
timestamp

sales pre-aggregation

DEV

db_dev

CREATE TABLE `DEV.db_dev.daily_sales_summary` (
  product_id BIGINT,
  date DATE,
  total_quantity BIGINT,

INSERT INTO `DEV.db_dev.daily_sales_summary`
SELECT 
  product_id,
  DATE_FORMAT(order_date, 'yyyy-MM-dd') AS date
  SUM(quantity) AS total_quantity,
  SUM(quantity * price) AS total_revenue
FROM 
  sales
GROUP BY 
  product_id, DATE_FORMAT(order_date, 'yyyy-MM-dd');

Results: >5k

Start: 2022-12-10 14:42:11

Duration: 820ms

Job Id: qewg-2t1t-xf43

SELECT * FROM daily_sales_summary LIMIT 10;

Results: 816

Start: 2022-12-10 14:51:32

Duration: 240ms

Job Id: xf43-qewg-2t1t

product_id BIGINT

date DATE

total_quantity BIGINT

total_revenue DECIMAL

101

2023-01-15

102

2023-01-15

101

2023-01-16

103

2023-01-16

104

2023-01-17

125

Stream lineage diagram showing data flow from source topics through transformations to sink

100%

Business Impact

Shift left to increase data flow through Confluent.

Confluent

Data warehouse

Application

Source Data Collection
Initial data is gathered from various sources, including user inputs, databases, external APIs.
Extract and Load
Data is extracted from the sources and loaded into a data warehouse for further processing.
Raw Data
The collected data is available in its raw, unprocessed form, ready for initial review and assessment.
Cleanse and Process
The raw data is cleansed and processed to remove errors, inconsistencies, and duplicates.
Business-Ready Data
The cleansed and processed data is structured and organized in a way that meets business requirements, ready for analysis.
Utilize Data
The data is used for various purposes, supporting decision-making and operational processes.
AI, Report, Dashboard
This data is used for advanced analytics, AI models, and report dashboard for actionable insights.

Flink and Kafka enable shift-left data processing, freeing data teams to focus on business priorities while Confluent grows its customer base and increases data flow.

Confluent

Data warehouse

Application

Source Data Collection
Initial data is gathered from various sources, including user inputs, databases, external APIs.
Raw Data
The collected data is available in its raw, unprocessed form, ready for initial review and assessment.
Cleanse and Process
The raw data is cleansed and processed to remove errors, inconsistencies, and duplicates.
Business-Ready Data
The cleansed and processed data is structured and organized in a way that meets business requirements, ready for analysis.
Extract and Load
Extraction and loading occur after data cleansing and processing, ensuring only business-ready data is loaded into the data warehouse.
Utilize Data
The data is used for various purposes, supporting decision-making and operational processes.
AI, Report, Dashboard
This data is used for advanced analytics, AI models, and report dashboard for actionable insights.

Build data pipeline

SQL Workspace to process data and explore schema — all in one place.

payments

Id
uuid
tripID
varchar
method
enum
amount
float
createdAt
timestamp

-- Retrieve basic trip details including customer and driver information

SELECT 
	customers.name AS customer_name,
	drivers.name AS driver_name
	trips.start_location,
	trips.end_location,
FROM 
	trips
JOIN 
	customers ON trips.customer_id = customers.customer_id
JOIN 
	drivers ON trips.driver_id = drivers.driver_id;

Results: >5k

Start: 2022-12-10 14:42:11

Duration: 450ms

Job Id: qewg-2t1t-xf43

customer_name varchar

driver_name varchar

start_location point

end_location point

123 Elm St

456 Oak St

John Doe

Jane Smith

789 Pine Ave

101 Maple Dr

Alice Johnson

Bob Brown

202 Birch Blvd

303 Cedar Lane

Michael Green

Sara White

404 Walnut St

505 Chestnut St

Emily Davis

Tom Black

606 Willow Way

707 Spruce St

David Wilson

Anna Lee

808 Poplar Pl

909 Redwood Rd

Sarah Connor

John Connor

123 Oakridge Ct

456 Highland Ave

Frank Castle

Karen Page

789 Summer St

101 Autumn Dr

Peter Parker

Tony Stark

202 Winter Blvd

303 Spring Ln

Bruce Wayne

Alfred Penny

404 Hero Rd

505 Villain Ave

Clark Kent

Lois Lane

606 Gotham St

707 Metropolis Blvd

Diana Prince

Steve Trevor

808 Lantern Dr

909 Arrow Ave

Barry Allen

Oliver Queen

123 Flash Blvd

456 Arrow Ct

Hal Jordan

Carol Ferris

789 Batman St

101 Robin Rd

Arthur Curry

Mera Anderson

202 Wonder Woman Ln

303 Aquaman Blvd

Victor Stone

Cyborg Thomas

Users constantly switch between the UI, where they explore data, and the CLI, where they write code and create streaming applications. SQL Workspace bridges both, providing a seamless development experience.

Understand data pipeline

Stream lineage and Query profiler make pipelines easier to discover, understand, and debug.

Stream lineage maps the pipeline, and the query profiler exposes task-level execution details, giving visibility into how a statement runs. This accelerates detection of bottlenecks, skew, and other performance problems.

Query profiler diagram showing pipeline stages from source through join, filter, calc to sink

Simplify development

Makes SQL easier to write and faster to get started.

-- Retrieve trip details including customer and driver info

SELECT 
  customers.name AS customer_name,
  drivers.name AS driver_name
  trips.start_location,
  trips.end_location,
FROM 
  trips
JOIN 
  customers ON trips.customer_id = customers.customer_id
JOIN 
  drivers ON trips.driver_id = drivers.driver_id;

CREATE TABLE `DEV.db_dev.table_name` (
  column1 TYPE,
  column2 TYPE,
  ...
)
AS SELECT id, name, email, createAt
FROM `DEV.db_dev.customers`
WHERE ...;

Artifact

cal_geolocation.jar

fo91-lfcp-359zm0

Environment

DEV

Cloud & Region

AWS | us-west-1 (Oregon)

SQL

USE DEV;
USE cluster_rides;

CREATE FUNCTION cal_geolocation AS 
  'com.example.cal_geolocation'
USING JAR
  'confluent-artifact://fo91-lfcp-359zm0';

User interviews revealed that despite similarities to regular SQL, users struggle with writing accurate Flink SQL queries. To address this, we introduced a Flink SQL AI bot and template features for quick insertion of common queries based on data configuration.

Agentic AI

Build real-time AI agents.

Flink offers a significant advantage for running Agentic AI with its real-time, event-driven capabilities, enabling enterprises to build scalable and reliable AI agents.

Agent configurations

Agent name

Instructions

You are an AI agent designed to assist a research team in processing and analyzing large-scale, complex datasets. Through iterative cycles, you continuously refine your outputs by evaluating intermediate results, identifying patterns, and adjusting computations to improve accuracy. Each iteration allows you to optimize the trade-offs between computational cost, latency, and prediction reliability, enabling the team to make data-driven decisions efficiently. Additionally, your iterative process supports adaptive learning, as feedback from prior outputs informs subsequent analyses, ensuring that insights become progressively more precise over time.

Tools

Allow agent to access and interact with external systems and data.

Model

Determine the intelligence, speed, and how to handle complex tasks.

Handle exception

How the agent should behave when an error occurs.

Max consecutive failures

Number of tool/step failures allowed before stopping

Max iterations

Maximum number of reasoning/acting loop cycles

Max token threshold

Token limit that triggers trimming or summarization

Request timeout

Maximum allowed runtime for each request

Summarization prompt

Style of summary used when compressing context.

Tokens management strategy

How the runtime reduces tokens when context gets too large.

{
  "id": 4729,
  "name": "Marble Finch",
  "active": true,
  "tags": ["alpha", "beta", "gamma"],
  "metadata": {
    "created_at": "2025-11-15T10:23:54Z",
    "version": 3
  },
  "scores": [12, 48, 7],
  "profile": {
    "role": "explorer",
    "level": 19
  }
}

2024-06-18 17:22:55

{
  "id": 4729,
  "name": "Marble Finch",
  "active": true,
  "tags": ["alpha", "beta", "gamma"],
  "metadata": {
    "created_at": "2025-11-15T10:23:54Z",
    "version": 3
  },
  "scores": [12, 48, 7],
  "profile": {
    "role": "explorer",
    "level": 19
  }
}

Flink actions

Make data processing accessible to a broader audience, not just data experts.

Create embeddings

Transform column data into vector embeddings and append to the topic.

De-duplicate topic

Generate a topic containing only unique records from an input topic.

Mask topic

Generate a topic containing masked fields from an input topic.

Transform topic

Apply custom transformations to the data in a topic.

Join topic

Combine records from multiple input topics to produce a unified output topic.

Filter topic

Select and retain only records that meet specific criteria from an input topic.

Transform topic

Apply custom transformations to the data in a topic.

Action details

customerId

uuid

driverId

uuid

paymentId

uuid

status

enum

source

float

destination

float

Partition key

Partition count

Serialization format

Output topic

Catalog

Database

Table

Flink statement

573e-01c0-a49b

Status

Running

Compute pool

default-pool

Duration

33s

Scaling status

Fine

Statement type

INSERT INTO

Start time

2024-06-21 12:42:33

SQL

INSERT INTO completed_rides
SELECT 
    driver_id, passenger_id, start_location,
    end_location, start_time, end_time, fare
FROM 
    rides_source
WHERE 
    ride_status = 'COMPLETED';

Lineage

Message behind

Message in

Message out

Research revealed that most users perform similar processes on stream data. To address this JTBD, we introduced Flink Actions, a low-code solution that lets users quickly execute common data processing and cleaning tasks.

Infrastructure

Establish a robust infrastructure that meets the team's unique needs.

Region

Availability

GCP.uswest.env-03638o

Running

lfcp-359zm0

Current CFU

Max CFU

Cloud & Region

GCP | Las Vegas (us-west4)

Account

Execute long-running Flink statements using a service account to enhance security and ensure controlled resource access.

Service account

Service account name

Service account description

pl-egress-01

Ready

gw-0df98fr

AWS

us-east-2

Egress

Network details

DNS domain

us-east2.aws.private.cflt.cloud

Connection type

PrivateLink gateway

PrivateLink ID

com.amazon.aws.vcpe.us-east2.vcpe-sve-09dw08dgdg09sdf8

Access points

Flink runs on AWS, Azure, and Google Cloud. Compute pools simplify resource management, service accounts secure long-running tasks, and private networks protect data across Confluent.

Confluent

Artifact

SQL

Agent configurations

Create embeddings

De-duplicate topic

Mask topic

Transform topic

Join topic

Filter topic

Transform topic

Action details

Output topic

573e-01c0-a49b

SQL

Lineage

GCP.uswest.env-03638o

Account

pl-egress-01

Network details

Access pointsCreate access point

Access points