Flink 2.2 ML_PREDICT and VECTOR_SEARCH: Running Model Inference Inside a Stream

Every Flink job carries an assumption nobody writes down: that an operator’s per-record work is bounded and roughly uniform. Checkpoint intervals, buffer sizing, watermark strategies and restart budgets are all tuned against that assumption. Flink 2.2 ML_PREDICT breaks it in one line of SQL, and VECTOR_SEARCH breaks it again a line later. The Apache Flink PMC shipped 2.2.0 on 4 December 2025 with 73 contributors, 9 FLIPs and over 220 resolved issues, and the headline was AI: large language model inference in SQL, vector similarity search in SQL. The syntax is genuinely clean. The operational physics underneath it are not.

This post is about the physics rather than the syntax. The reference docs already show you how to spell the function call. What they do not tell you is what happens to backpressure, checkpoint state, end-to-end latency, per-record cost and failure semantics once an unbounded, high-variance, rate-limited remote dependency sits on the hot path of a stateful streaming job.

What this covers: what Flink 2.2 actually shipped and the exact confirmed SQL surface; how the async operator interacts with barriers and watermarks; a worked cost-per-million-records model; the specific failure modes that produce restart loops; and a decision rule for when an async sink plus an external inference service is the better design.

Context and Background

Calling a model from a stream is not new. Teams have been doing it since the first AsyncFunction shipped, usually as a hand-written DataStream operator wrapping an HTTP client, with retry, caching and concurrency control written by hand and reviewed by nobody. The novelty in 2026 is that the call is now a first-class SQL construct, which means it can be written by people who have never read Flink’s async I/O documentation.

The lineage is short. FLIP-437 introduced models as catalog objects in Flink SQL, and FLIP-525 defined the implementation design for ML_PREDICT and ML_EVALUATE. Those landed in Flink 2.1 (released 31 July 2025), which the 2.2 release notes describe as “technically validated in scenarios such as log classification and real-time question-answering systems.” Flink 2.2 then extends model inference to the Table API through FLIP-526 and FLINK-38104, and adds VECTOR_SEARCH through FLIP-540 and FLINK-38422.

That second addition is the structurally interesting one. Before 2.2, the release blog notes, Flink could only “use embedding models to convert unstructured data into high-dimensional vector features, which are then persisted to downstream storage systems.” Retrieval had to happen somewhere else. With VECTOR_SEARCH, retrieval-augmented generation becomes expressible end to end inside a single SQL statement: embed, retrieve, generate.

Commercially, Confluent and Ververica shipped managed variants of the same idea ahead of upstream, and Alibaba Cloud’s Realtime Compute service documents its own ML_PREDICT. That matters when you read blog posts: the managed dialects have diverged from Apache Flink on options, providers and even argument names. Always check which dialect a snippet targets before pasting it. The same care applies when you are choosing the engine at all — the trade-off space differs sharply across engines, as covered in our Flink vs Spark Structured Streaming vs Kafka Streams comparison.

The rest of Flink 2.2 is unusually relevant to this feature even though it is not marketed alongside it: balanced task scheduling (FLIP-370), a RateLimiter for scan sources (FLIP-535), skew-aware split assignment (FLIP-537), SinkUpsertMaterializer V2 (FLIP-544), and finer-grained checkpoint tracing via traces.checkpoint.span-detail-level. Several of those are the levers you will reach for when inference starts distorting your job.

The Core Argument: A Remote Call Inside an Operator Is a New Latency Class

ML_PREDICT is a Flink SQL table-valued function that takes a table, a registered model and a descriptor of feature columns, and returns the input table plus the model’s output columns. VECTOR_SEARCH is a table-valued function that takes a vector table, a query vector column, an index descriptor and a top_k, and returns the matched rows plus a score column. Both run as lookup-style operators against an external system.

Figure 1: A real-time RAG pipeline expressed entirely in Flink SQL. Every double-headed edge is a network round trip on the record path.

The figure shows the shape that the 2.2 feature set makes possible: a Kafka source feeding a filter, an ML_PREDICT call against an embedding model, a VECTOR_SEARCH against an external vector store, a second ML_PREDICT against a generation endpoint, and a sink. Note what the arrows back from the vector store and the model endpoint represent. Each is a synchronous dependency on a system Flink does not schedule, does not checkpoint, and cannot backpressure. Three of them sit in series on the same record.

The confirmed SQL surface

Getting the signatures right matters more than usual here, because the shipped syntax differs from the original FLIP text in at least one place. The following is from the Flink 2.2 documentation.

Model registration uses CREATE MODEL with explicit input and output schemas:

CREATE MODEL ai_analyze_sentiment
INPUT (`input` STRING)
OUTPUT (`content` STRING)
WITH (
  'provider'     = 'openai',
  'endpoint'     = 'https://api.openai.com/v1/chat/completions',
  'api-key'      = '<YOUR KEY>',
  'model'        = 'gpt-3.5-turbo',
  'system-prompt' = 'Classify the text below into one of the following labels: [positive, negative, neutral, mixed]. Output only the label.'
);

Inference is a table-valued function with an optional configuration map:

-- positional form
SELECT *
FROM ML_PREDICT(
  TABLE input_table,
  MODEL my_model,
  DESCRIPTOR(feature1, feature2),
  MAP['async', 'true', 'timeout', '100s']
);

-- named-parameter form
SELECT *
FROM ML_PREDICT(
  INPUT  => TABLE input_table,
  MODEL  => MODEL my_model,
  ARGS   => DESCRIPTOR(feature1, feature2),
  CONFIG => MAP['async', 'true']
);

Vector search is a lateral table function:

SELECT *
FROM input_table, LATERAL TABLE (VECTOR_SEARCH(
  TABLE vector_table,
  input_table.vector_column,
  DESCRIPTOR(index_column),
  10,
  MAP['async', 'true', 'timeout', '100s']
));

Use the named-parameter form for VECTOR_SEARCH in production code: SEARCH_TABLE, COLUMN_TO_QUERY, COLUMN_TO_SEARCH, TOP_K, CONFIG. FLIP-540 proposed the positional order as search table, column-to-search, column-to-query, top_k; the shipped documentation shows search table, query column, index descriptor, top_k. Named parameters remove the ambiguity entirely, and they survive a future reordering.

Four configuration keys are shared by both functions, and they are the only runtime knobs you get: async (Boolean), max-concurrent-operations (Integer), output-mode (ORDERED or ALLOW_UNORDERED), and timeout (Duration). All four default to unset. The output-mode description in the docs explicitly says the value converts to AsyncDataStream.OutputMode, which tells you the async path is Flink’s ordinary async I/O machinery with a SQL wrapper on top. That single sentence is the most useful thing in the reference page, because it means everything already known about AsyncWaitOperator applies.

Sync and async are two different operators

A model provider can implement PredictRuntimeProvider for synchronous inference, AsyncPredictRuntimeProvider for asynchronous inference, or both. If async is not set in the config map, the planner picks one and prefers the async provider when both exist. If you request a mode the provider does not support, the job fails at planning time with an explicit error rather than silently degrading.

This is a bigger decision than it looks. The synchronous path blocks the operator’s task thread for the duration of the call. At 300 ms per call, one subtask sustains roughly three records per second. Reaching 5,000 records per second would need a parallelism in the thousands, which is absurd. In practice, synchronous mode is only defensible for local or co-located models with sub-millisecond latency, or for very low-rate control streams. Everything else needs async = true, and therefore needs to reason about concurrency.

Concurrency is Little’s Law, and nothing else

max-concurrent-operations is the maximum number of async I/O operations a subtask may have outstanding. Total in-flight concurrency for the job is that value multiplied by operator parallelism. Sizing it is Little’s Law: the number of concurrent requests you must sustain equals arrival rate times service time.

At 5,000 records per second with a 300 ms median endpoint latency, you need 1,500 concurrent requests just to keep up at the median. If the endpoint’s p99 is 2 seconds, absorbing the tail without stalling needs headroom toward 10,000. With a parallelism of 32, that is roughly 47 in flight per subtask at the median and around 310 per subtask to cover the tail. Set max-concurrent-operations below the first number and the job backpressures immediately; set it near the second and every checkpoint has to snapshot up to 10,000 buffered records across the job.

Neither number is available from the Flink UI’s usual backpressure indicators, because the operator is not CPU-busy — it is waiting. This is the first practical consequence of the thesis: the normal diagnostic vocabulary for a Flink job (“busy time”, “backpressured time”, “records in per second”) describes an operator whose latency is bounded. An inference operator sitting at 3% busy time and 100% of its concurrency budget looks idle and is in fact saturated.

Deeper Analysis: Barriers, Watermarks and the Cost Model

Three mechanisms determine whether inference-in-stream works at your volume. Two are about correctness machinery; one is about money. All three are quantifiable in advance, which is the point of doing this analysis before you ship.

Figure 2: What a checkpoint barrier actually does when requests are in flight — and what recovery does to the endpoint.

The sequence in Figure 2 traces a barrier from the JobManager through the source, into the ML_PREDICT operator while requests are outstanding, into the snapshot, and then through a committing sink. The final two steps are the ones that cause incidents: on restore, the operator re-fires every request that was pending when the checkpoint was taken.

Checkpoints: the barrier is not blocked, but the state and the recovery are

A common assumption is that a barrier cannot pass an async operator with outstanding requests. It can. The Flink async I/O documentation is explicit: the operator “offers full exactly-once fault tolerance guarantees. It stores the records for in-flight asynchronous requests in checkpoints and restores/re-triggers the requests when recovering from a failure.”

That is good news for checkpoint duration and bad news in three other ways.

First, checkpoint size becomes a function of endpoint latency. In-flight records are operator state. At 10,000 in-flight records averaging 4 KB of payload, that is 40 MB of extra state per checkpoint, written every checkpoint interval, purely because the endpoint is slow. Unaligned checkpoints do not help; this is operator state, not in-flight network buffers.

Second, recovery is a thundering herd. Restoring from a checkpoint re-triggers every pending request simultaneously. A job that was comfortably issuing 5,000 requests per second now issues 10,000 in a burst at t=0. If the endpoint enforces a quota, that burst produces a wall of HTTP 429s, which the provider retries, which extends latency, which can trip the job’s timeout, which fails the job, which restores from the same checkpoint and re-fires the same burst. This is a genuine restart loop and it is self-reinforcing.

Third — and this is the part that bites analytically rather than operationally — Flink’s exactly-once is exactly-once with respect to Flink state, not with respect to the endpoint. Re-fired requests are billed again. And because the docs state plainly that “ML_PREDICT results are non-deterministic” (the reason CDC input tables are rejected and only append-only tables are supported), the answer you get after recovery may differ from the one you got before the failure. Any downstream aggregate computed over those labels is therefore not reproducible across a restart. If you have compliance requirements around reproducibility, this alone disqualifies in-stream inference.

Watermarks: why ALLOW_UNORDERED often buys you nothing

output-mode defaults to ORDERED. ALLOW_UNORDERED is documented as attempting unordered emission “when it does not affect the correctness of the result, otherwise ORDERED will be still used.” The condition hiding in that sentence is event time.

Flink’s async I/O documentation spells out the semantics. Under event time, “watermarks do not overtake records and vice versa, meaning watermarks establish an order boundary. Records are emitted unordered only between watermarks.” And then, bluntly: “in the presence of watermarks, the unordered mode introduces some of the same latency and management overhead as the ordered mode does. The amount of that overhead depends on the watermark frequency.”

Quantify that. pipeline.auto-watermark-interval defaults to 200 ms. At 5,000 records per second, each 200 ms watermark epoch contains about 1,000 records. Before the watermark can be emitted, every result for those 1,000 records must be emitted first. The epoch therefore closes at the maximum latency among 1,000 draws from the endpoint’s latency distribution — and the maximum of 1,000 samples from a distribution with a 2-second p99 will almost always be at least 2 seconds.

So each 200 ms of event time costs roughly 2 seconds of wall clock. The job progresses at about a tenth of real time and falls behind permanently. This is an order-of-magnitude argument, not a simulation, and per-partition watermark alignment softens it somewhat — but the direction is right and the remedy is clear. Either lengthen the watermark interval by an order of magnitude (accepting coarser window firing), or run the inference stage on processing time, or take the inference off the record path entirely.

Practitioners who have tuned lookup joins against slow external stores will recognise this shape. The difference is magnitude: a well-run key-value lookup has a p99 in single-digit milliseconds, while a generation endpoint’s p99 is measured in seconds. The same architecture pattern moves three orders of magnitude and stops working.

Chained hops multiply tail exposure

Figure 1 has three remote hops per record: embed, retrieve, generate. If each hop independently exceeds its p99 threshold 1% of the time, the probability that at least one hop in the chain is a tail event is 1 − 0.99³ ≈ 2.97%. The chain’s p97 is therefore worse than any single hop’s p99. Latency budgets built per-hop systematically understate end-to-end behaviour, and the understatement grows with chain length.

This is a good argument for collapsing hops. If your vector store and your generation endpoint are both remote, consider whether retrieval can be served from Flink state (a broadcast or temporal table of embeddings) rather than a network call. FLIP-540 explicitly scopes 2.2 to “leveraging an external system for vector search based on processing time” and defers the state-backed and event-time implementations to future FLIPs — so today that collapse has to be hand-built.

Cost per million records: a worked example

Vendor prices change monthly, so treat the rates below as parameters and substitute your provider’s current published numbers. The structure of the calculation is what matters.

Take a classification pipeline: 5,000 records per second, each record producing a prompt of roughly 250 input tokens and 8 output tokens (a single label). Assume an illustrative $0.40 per million input tokens and $1.60 per million output tokens.

Quantity	Value
Input tokens per million records	250,000,000
Output tokens per million records	8,000,000
Input cost per million records	$100.00
Output cost per million records	$12.80
Total per million records	$112.80
Records per day at 5,000/s	432,000,000
Inference cost per day	$48,729
Inference cost per 30 days	$1,461,888

Now price the Flink side. Eight TaskManager instances at an illustrative $0.35 per hour is $2.80 per hour, or about $67 per day. Inference is roughly 700× the cost of the stream processing that hosts it. That ratio is the single most important number in this entire design space, and it inverts every instinct a data engineer has about where to spend optimisation effort.

The levers, in descending order of impact:

Filter before you predict. If only 3% of records genuinely need a model decision, a WHERE clause ahead of ML_PREDICT cuts the bill from $48,729 to about $1,462 per day. Flink’s planner will not push a filter through a table-valued function for you in every case — write the filter as an explicit subquery or view and verify with EXPLAIN.

Shorten the prompt. Input tokens dominate at 89% of the cost here. Cutting a 250-token prompt to 90 tokens by removing boilerplate and few-shot examples saves about $64 per million records, or $27,600 per day, for a morning’s prompt engineering.

Deduplicate. Telemetry and log streams are extremely repetitive. Keying by a content hash and maintaining a Flink state-backed cache of recent verdicts routinely removes 60–90% of calls on log classification workloads. There is no built-in result cache in ML_PREDICT; you build this yourself with a keyed state lookup before the predict step.

Pick a smaller model per stage. Routing with a cheap classifier and escalating only ambiguous records to the expensive model is a two-stage cascade that typically preserves most of the accuracy at a fraction of the cost.

Batch. This is the lever ML_PREDICT does not give you. The function is per-record; there is no documented micro-batching across records. Modern serving stacks get most of their throughput from continuous batching, and you forfeit that by calling the endpoint one record at a time. The economics of that forfeit are covered in our vLLM cost economics deep dive, and it is the strongest structural argument for moving inference out of the operator.

Rate Limits, Retries and Partial Failure

The Apache Flink OpenAI model connector exposes a specific error-handling contract, and its defaults are not what you would choose for a production stream.

Figure 3: The documented error path for the OpenAI model provider, including the recovery edge that re-fires pending requests.

error-handling-strategy defaults to RETRY, with the alternatives FAILOVER (throw and fail the job) and IGNORE (skip the offending input and log the error). retry-num defaults to 100. retry-fallback-strategy, which applies once retries are exhausted, defaults to FAILOVER.

Read those three defaults together. A record that hits a sustained 429 will be retried up to 100 times before the job fails. With any realistic backoff, 100 retries is minutes of wall clock during which that async slot is occupied and timeout — which is unset by default — never fires. One poisoned key can hold a concurrency slot hostage for minutes, and a rate-limit event that affects all keys will hold all of them.

There is a second-order effect. The timeout option is documented as measuring “from first invoke to final completion of asynchronous operation, may include multiple retries, and will be reset in case of failover.” The retry budget lives inside the timeout window, so setting timeout is the only way to bound the retry storm. And the reset-on-failover clause means a job that keeps failing over keeps getting a fresh timeout budget — another ingredient of the restart loop.

The escape hatch is error-handling-strategy = 'IGNORE' combined with the provider’s error metadata columns. When ignoring errors, you can declare metadata columns in the model’s output schema to surface failure detail into the stream: error-string (STRING), http-status-code (INT), and http-headers-map (MAP of STRING to ARRAY). If the call succeeds, those columns are null. That turns a fail-stop dependency into a nullable column you can route, count and alert on — which is how a streaming job should treat any external dependency.

Context handling deserves a mention because it fails silently. max-context-size combines with context-overflow-action, which defaults to truncated-tail: tokens beyond the limit are cut from the end of the context and, by default, nothing is logged. A schema change upstream that lengthens a field will start silently truncating prompts, and your classification accuracy will drift with no error anywhere. Use the -log variants (truncated-tail-log, truncated-head-log, skipped-log) in production so truncation is observable.

One more rate-limiting trap. Flink 2.2 introduced a RateLimiter interface for scan sources through FLIP-535, and it is tempting to assume that solves endpoint throttling. It does not. It is source-side, it is currently DataStream API only, and it limits how fast you read, not how fast you call the model. The only throttle on the inference path is max-concurrent-operations, which limits concurrency rather than rate. Concurrency limiting and rate limiting are different controls: with a concurrency cap of 1,500 and a service time that suddenly drops from 300 ms to 30 ms, your request rate jumps tenfold and blows the quota you thought you were respecting.

Trade-offs, Gotchas, and What Goes Wrong

The honest summary is that ML_PREDICT is excellent for a narrow band of workloads and actively hazardous outside it. Here is the specific list.

Append-only only, and for a reason. Both ML_PREDICT and VECTOR_SEARCH accept append-only tables exclusively. The docs give the reason for ML_PREDICT: results are non-deterministic, so applying it to a changelog would produce retractions that disagree with the rows they retract. This is correct behaviour, but it means you cannot drop inference into the middle of an existing upsert pipeline without restructuring it.

Column name collisions are silently renamed. If a model’s output column is prediction and the input table already has a prediction column, the output is renamed to prediction0. Downstream SELECT * consumers will not notice until a schema validation somewhere else rejects the row. Name model output columns defensively.

VECTOR_SEARCH reads the latest snapshot. The function “uses a processing-time attribute to correlate rows to the latest version of data in an external table.” Re-running yesterday’s data against today’s index gives different results. FLIP-540 proposed an optional ON_TIME argument for event-time snapshots but scoped it out of the initial implementation, so there is currently no way to pin retrieval to the record’s event time. Replay is not reproducible.

Tie-breaking is first-come-first-served. FLIP-540 states that when more rows share a score than top_k allows, the behaviour follows TOP-N — arbitrary among ties — rather than RANK or DENSE_RANK. Results at the boundary are not stable across runs.

NULL query vectors behave like a join miss. A NULL input vector matches nothing. With LEFT JOIN LATERAL you get one row with nulls; with inner lateral the row is filtered out entirely and disappears from your stream. Silent row loss from a nullable embedding column is a genuinely hard bug to find.

You must supply the vector source. The vector table has to implement org.apache.flink.table.connector.source.VectorSearchTableSource. That is a small connector ecosystem today. Check your store has a maintained implementation before designing around it.

The LATERAL keyword is conditional. It is required when the search correlates with the input table and may be omitted when the query vector is a constant. Both forms appear in the official examples, which is a reliable source of copy-paste errors.

Restart budget interacts with everything above. Given re-fired requests on recovery, a restart strategy with a short delay makes the thundering herd worse. Pair inference jobs with a longer restart delay and exponential backoff than you would use for a pure-compute job.

When an Async Sink Plus an External Service Wins

Figure 4: Four questions that decide whether inference belongs inside the operator or behind a request topic.

The alternative design is old and boring: Flink writes an inference-request record to a topic through an async sink, a separate consumer pool of GPU workers does continuous batching and publishes results to a results topic, and — only if the enriched result is actually needed in the streaming job — a second Flink job joins the results back via a lookup join, interval join or delta join.

It costs an extra hop and adds tens of milliseconds of floor latency. In exchange you get four things ML_PREDICT cannot give you.

Independent scaling. The GPU pool scales on queue depth; the Flink job scales on partition count. Neither forces a restart of the other. Deploying a new model version does not touch the streaming topology, and a bad model rollout does not fail a stateful job.

Batching. Continuous batching is where serving throughput comes from. A request topic gives the worker pool a natural batch boundary and lets it choose batch size dynamically against GPU memory. This routinely improves tokens-per-dollar several-fold over per-record calls.

Bounded operator latency. The async sink write is a local Kafka produce with a p99 in single-digit milliseconds. Your operator latency distribution goes back to being bounded, which means your watermark strategy, checkpoint interval and backpressure diagnostics all start telling the truth again. This is the thesis stated in reverse: the fix is not tuning the inference call, it is removing it from the correctness-critical path.

Natural durable buffering. The request topic absorbs endpoint outages. A three-minute provider incident becomes lag rather than a restart loop, and the retention window is your replay buffer. Getting the buffer topic right — partitioning, retention, ordering — follows the same rules as any ingest bridge, which our MQTT to Kafka bridge production tutorial works through in detail, and the broker choice itself is covered in the Kafka vs Redpanda vs WarpStream ADR.

Conversely, keep inference inside the job when all four of these hold: throughput is in the hundreds of records per second rather than thousands; the model verdict is needed immediately to route, filter, join or window within the same job; the endpoint is co-located with a p99 well under 200 milliseconds; and reproducibility across replay is not a requirement. Embedding generation against a local model and VECTOR_SEARCH against a low-latency vector database both sit comfortably inside that envelope. Chat-completion calls against a public API at five-figure QPS do not.

Practical Recommendations

Start by measuring, not building. Before writing any SQL, collect your endpoint’s latency distribution at your intended concurrency — p50, p99 and p99.9 — and your quota in requests per second and tokens per minute. Every decision below falls out of those five numbers.

Then size for the tail rather than the median. Compute concurrency from Little’s Law at p99, divide by parallelism, and set max-concurrent-operations to that per-subtask figure. Multiply back out to get the in-flight record count, multiply by average record size, and confirm that the resulting checkpoint state increase is acceptable at your checkpoint interval.

Treat the model endpoint as a nullable column, not as a dependency that may fail the job. Configure error-handling-strategy = 'IGNORE', declare the error metadata columns, and route failures to a dead-letter stream with alerting on the failure rate. A job that stops because an API returned 503 is a job that will stop weekly.

A pre-deployment checklist:

• [ ] Set timeout explicitly — the default is unset and the 100-retry default lives inside it.
• [ ] Set retry-num to single digits and retry-fallback-strategy = 'IGNORE'.
• [ ] Use the -log variants of context-overflow-action so truncation is visible.
• [ ] Put a filter and a state-backed dedup cache in front of ML_PREDICT; verify with EXPLAIN.
• [ ] Raise pipeline.auto-watermark-interval or move the inference stage to processing time.
• [ ] Use named parameters for VECTOR_SEARCH; never rely on positional order.
• [ ] Alias model output columns explicitly to avoid silent prediction0 renames.
• [ ] Lengthen the restart delay to blunt the re-fire burst on recovery.
• [ ] Track cost per million records as a first-class SLO next to lag and checkpoint duration.
• [ ] Enable traces.checkpoint.span-detail-level so you can see which operator owns checkpoint time.

Frequently Asked Questions

What is ML_PREDICT in Flink SQL?

ML_PREDICT is a table-valued function that applies a registered model to a streaming table. You pass it the input table, a model created with CREATE MODEL, and a DESCRIPTOR listing which columns are features. It returns every input column plus the model’s output columns. It supports both synchronous and asynchronous execution depending on the provider, accepts an optional configuration map, and works only on append-only tables because model results are non-deterministic.

Is ML_PREDICT new in Flink 2.2?

No. ML_PREDICT arrived in Flink SQL in version 2.1 via FLIP-437 and FLIP-525. Flink 2.2, released 4 December 2025, extends model inference to the Table API through FLIP-526 and adds the genuinely new VECTOR_SEARCH function through FLIP-540. The 2.2 release also ships balanced task scheduling, a source-side rate limiter, delta join improvements and PyFlink async function support, several of which matter when tuning inference workloads.

How does VECTOR_SEARCH differ from a lookup join?

Mechanically they are close cousins: both correlate an input row against the latest snapshot of an external table using a processing-time attribute. The difference is the matching predicate. A lookup join matches on key equality; VECTOR_SEARCH compares a query vector against an indexed vector column and returns the top_k most similar rows plus a score column. The vector table must implement VectorSearchTableSource, and only append-only input is supported.

Does ML_PREDICT block Flink checkpoints?

Not directly. Flink’s async I/O operator snapshots in-flight requests as operator state, so the barrier is not held waiting for the endpoint. The real costs are elsewhere: checkpoint state grows in proportion to in-flight concurrency, and on recovery every pending request is re-triggered at once, which can overwhelm a rate-limited endpoint and cause a restart loop. Exactly-once applies to Flink state, not to endpoint billing.

How much does streaming LLM inference cost?

Model it per million records rather than per request. At 250 input and 8 output tokens per record, and illustrative rates of $0.40 and $1.60 per million tokens, one million records costs about $113 — roughly $48,700 per day at 5,000 records per second. The Flink cluster hosting that is a tiny fraction of it. Filtering, prompt shortening, deduplication and model cascading are far higher-leverage than any cluster tuning.

When should inference stay outside the Flink job?

Move it out when throughput is in the thousands of records per second, when you need continuous batching for GPU efficiency, when the endpoint’s p99 exceeds a few hundred milliseconds, when replay must be reproducible, or when the model’s deployment lifecycle should be independent of the streaming topology. The standard pattern is an async sink to a request topic, a GPU worker pool doing dynamic batching, and a join back only if the stream genuinely needs the result.

Further Reading

• Flink vs Spark Structured Streaming vs Kafka Streams: a 2026 comparison — engine selection before you design the inference path.
• Kafka vs Redpanda vs WarpStream for edge telemetry: an architecture decision record — choosing the broker that will hold your inference request topic.
• vLLM cost economics deep dive — why continuous batching dominates serving cost, and what per-record calls give up.
• Building an MQTT to Kafka bridge with Kafka Connect — the ingest half of the same pipeline.
• Apache Flink 2.2.0 release announcement — the primary source for every 2.2 claim above.
• Flink SQL model inference reference and VECTOR_SEARCH reference — the confirmed signatures and configuration options.

By Riju — about