Definition
- DataStream API – A low-level, imperative programming interface where you process events one at a time and have full explicit control over state handles, event-time and processing-time timers, windows, and custom operator logic.
- Table API – A high-level, declarative interface where you express computations as relational operations on tables using a fluent Java or Scala API.
- Flink SQL – A declarative SQL interface where you write standard SQL queries over streams and batch tables, with Flink's cost-based optimizer planning execution.
- Unified batch and streaming – The Table API and Flink SQL treat bounded and unbounded data uniformly, so one query serves both batch and streaming.
- Interoperability – You can convert a Table to a DataStream and back within one job, using each API where it is strongest.
Imagine the National Bank of India needs to process transactions. The DataStream API is like giving a skilled operations officer a detailed manual to follow for each individual transaction, where they decide exactly what to remember, when to set a reminder timer, and how to handle every edge case by hand. Flink SQL is like handing the same work to an analyst who writes a plain request: show me the total fraud-flagged amount per account per five-minute window. A smart system figures out the most efficient way to compute it. For straightforward aggregations the analyst's declarative request is far quicker; for intricate custom decision logic the officer's manual gives control the analyst's language cannot express.
Understanding the Concepts
What is the DataStream API?
The DataStream API is Flink's foundational interface. You work directly with streams of events and write functions that process each event. With a KeyedProcessFunction you get explicit access to keyed state through state handles like ValueState and ListState, and to event-time and processing-time timers that you register and fire yourself.
package com.bank.streaming;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import java.math.BigDecimal;
public class AccountBalanceTracker
extends KeyedProcessFunction<String, Transaction, String> {
private transient ValueState<BigDecimal> balance;
@Override
public void open(Configuration parameters) {
balance = getRuntimeContext().getState(
new ValueStateDescriptor<>("balance", BigDecimal.class));
}
@Override
public void processElement(Transaction txn, Context ctx,
Collector<String> out) throws Exception {
BigDecimal current = balance.value();
if (current == null) current = BigDecimal.ZERO;
BigDecimal updated = current.add(txn.getAmount());
balance.update(updated);
if (updated.compareTo(new BigDecimal("1000000")) > 0) {
out.collect("HIGH BALANCE ALERT: " + ctx.getCurrentKey()
+ " balance=Rs" + updated);
}
}
}
What is the Table API and Flink SQL?
The Table API and Flink SQL sit on top of the DataStream layer. You register streams and tables and then express transformations as relational operations. Flink's cost-based optimizer plans the query, pushes down filters, and manages state for stateful operations like joins and aggregations automatically.
package com.bank.streaming;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
public class FlinkSqlDemo {
public static void main(String[] args) {
TableEnvironment tEnv = TableEnvironment.create(
EnvironmentSettings.inStreamingMode());
tEnv.executeSql(
"CREATE TABLE nbi_transactions ("
+ " account_id STRING,"
+ " amount DECIMAL(15,2),"
+ " txn_time TIMESTAMP(3),"
+ " WATERMARK FOR txn_time AS txn_time - INTERVAL '5' SECOND"
+ ") WITH ("
+ " 'connector' = 'kafka',"
+ " 'topic' = 'nbi-transactions',"
+ " 'format' = 'json'"
+ ")");
// Windowed aggregation in pure SQL
tEnv.executeSql(
"SELECT account_id,"
+ " TUMBLE_START(txn_time, INTERVAL '5' MINUTE) AS window_start,"
+ " SUM(amount) AS total_amount,"
+ " COUNT(*) AS txn_count"
+ " FROM nbi_transactions"
+ " GROUP BY account_id,"
+ " TUMBLE(txn_time, INTERVAL '5' MINUTE)")
.print();
}
}
+----------+---------------------+--------------+-----------+
|account_id| window_start| total_amount| txn_count|
+----------+---------------------+--------------+-----------+
| ACC-001|2024-01-15 10:00:00.0| 125000.00| 12|
| ACC-002|2024-01-15 10:00:00.0| 45000.00| 5|
+----------+---------------------+--------------+-----------+
How is the DataStream API different from Flink SQL?
| Feature | DataStream API | Table API / Flink SQL |
|---|---|---|
| Paradigm | Imperative, event-at-a-time | Declarative, relational |
| State management | Explicit state handles (ValueState, ListState) | Automatic, managed by optimizer |
| Timers | Explicit registration and firing | Implicit via window definitions |
| Optimization | Manual operator tuning | Cost-based optimizer |
| Batch + streaming | Separate DataSet API (legacy) | Unified: same query for both |
| Code volume | Verbose, full control | Concise, less boilerplate |
| Target user | Streaming engineers | Analysts and engineers |
| Custom logic | Unlimited flexibility | UDFs for custom bits |
// DataStream API: verbose but full control
stream.keyBy(Transaction::getAccountId)
.window(TumblingEventTimeWindows.of(Time.minutes(5)))
.aggregate(new TransactionAggregator())
.process(new CustomAlertLogic());
// Flink SQL: concise, optimizer-managed
// SELECT account_id, SUM(amount) FROM txns
// GROUP BY account_id, TUMBLE(txn_time, INTERVAL '5' MINUTE)
How do you convert between DataStream and Table?
The two APIs are fully interoperable. You convert a DataStream to a Table, run SQL on it, then convert back to a DataStream when you need custom low-level logic. This lets you use each API where it is strongest within a single job.
package com.bank.streaming;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
public class InteropDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
// Start with DataStream
DataStream<Transaction> txnStream = env
.addSource(new KafkaSource<>("nbi-transactions"));
// Convert to Table for SQL aggregation
Table txnTable = tEnv.fromDataStream(txnStream);
tEnv.createTemporaryView("txns", txnTable);
Table aggregated = tEnv.sqlQuery(
"SELECT account_id, SUM(amount) AS total"
+ " FROM txns GROUP BY account_id");
// Convert back to DataStream for custom logic
DataStream<AccountTotal> resultStream =
tEnv.toDataStream(aggregated, AccountTotal.class);
resultStream
.keyBy(AccountTotal::getAccountId)
.process(new CustomFraudRuleEngine()) // low-level control
.addSink(new AlertSink());
env.execute("NBI Interop Demo");
}
}
What is Flink CEP and how does it relate?
Flink CEP (Complex Event Processing) is a pattern-matching library built on the DataStream API. It lets you define temporal patterns over event streams and detect matches. For example, detecting three transactions from the same account within thirty seconds where the third exceeds a limit.
package com.bank.fraud;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.windowing.time.Time;
public class CepFraudDemo {
public static void main(String[] args) throws Exception {
// Pattern: 3 transactions within 30 seconds, third > Rs50000
Pattern<Transaction, ?> fraudPattern = Pattern
.<Transaction>begin("first")
.next("second")
.next("third").where(
new SimpleCondition<Transaction>() {
@Override
public boolean filter(Transaction txn) {
return txn.getAmount()
.compareTo(new BigDecimal("50000")) > 0;
}
})
.within(Time.seconds(30));
PatternStream<Transaction> patternStream = CEP.pattern(
txnStream.keyBy(Transaction::getAccountId),
fraudPattern);
patternStream.select(new FraudAlertSelector())
.addSink(new AlertSink());
}
}
What are common mistakes when choosing between the APIs?
The most common mistake is writing everything in the DataStream API when most of the pipeline is straightforward relational logic that Flink SQL handles more concisely and with optimizer benefits. Another mistake is trying to force complex stateful timer logic into SQL when it would be simpler in a KeyedProcessFunction.
// WRONG: writing simple aggregation in DataStream API
// Verbose, no optimizer benefit
stream.keyBy(Transaction::getAccountId)
.window(TumblingEventTimeWindows.of(Time.minutes(5)))
.aggregate(new AggregateFunction<Transaction, Accumulator, Result>() {
// 30+ lines of boilerplate for a simple SUM
});
// CORRECT: use Flink SQL for simple aggregation
// SELECT account_id, SUM(amount)
// FROM txns
// GROUP BY account_id, TUMBLE(txn_time, INTERVAL '5' MINUTE)
When should you use each API?
Use the Table API or Flink SQL as the default for windowed aggregations, joins, filtering, projections, and any relational transformation. Drop to the DataStream API for custom event-time state machines, fine-grained timer logic, complex process functions, and integration with external systems in bespoke ways. Use the interop to combine both within one job.
Decision guide:
Windowed aggregation? --> Flink SQL
Stream-stream join? --> Flink SQL
Simple filter + project? --> Flink SQL
Custom timeout state machine --> DataStream (KeyedProcessFunction)
Complex event patterns? --> DataStream (Flink CEP)
Timer-driven business rules? --> DataStream (ProcessFunction)
Mix of both? --> Convert between Table and DataStream
What should a senior engineer know about the API choice?
A senior engineer should know that Flink SQL's optimizer automatically manages state for stateful operations like joins, that the Table API supports user-defined functions (UDFs, UDAFs, UDTFs) for custom logic within SQL, and that the cost-based optimizer can push down filters and predicates to connectors, reducing the data read. They should also know that Flink SQL's continuous query semantics mean that a streaming GROUP BY emits updates (changelog stream) not just inserts.
package com.bank.streaming;
import org.apache.flink.table.functions.ScalarFunction;
// Custom UDF for use in Flink SQL
public class FraudScoreUdf extends ScalarFunction {
public double eval(String accountId, BigDecimal amount) {
// Custom fraud scoring logic usable in SQL
if (amount.compareTo(new BigDecimal("100000")) > 0) {
return 0.9; // high risk
}
return 0.1; // low risk
}
}
// Register and use in SQL:
// tEnv.createTemporaryFunction("fraud_score", FraudScoreUdf.class);
// SELECT account_id, amount, fraud_score(account_id, amount) AS risk
// FROM nbi_transactions
// WHERE fraud_score(account_id, amount) > 0.5
Visual Overview
┌──────────────────────────────────────────────────┐
│ Flink API Layers │
│ │
│ ┌──────────────────────────────────────────────┐ │
│ │ Flink SQL / Table API │ │
│ │ Declarative, optimizer-planned, unified │ │
│ │ SELECT ... FROM ... GROUP BY ... │ │
│ └────────────────────┬─────────────────────────┘ │
│ │ converts to │
│ ┌────────────────────▼─────────────────────────┐ │
│ │ DataStream API │ │
│ │ Imperative, full control, state + timers │ │
│ │ stream.keyBy().process(new MyFunction()) │ │
│ └────────────────────┬─────────────────────────┘ │
│ │ executes on │
│ ┌────────────────────▼─────────────────────────┐ │
│ │ Flink Runtime Engine │ │
│ │ JobManager, TaskManagers, Slots, Checkpoints │ │
│ └──────────────────────────────────────────────┘ │
└──────────────────────────────────────────────────┘
Interop: Combining Both APIs in One Job
Kafka ──▶ DataStream ──▶ toTable() ──▶ SQL Aggregation
│
toDataStream()
│
▼
KeyedProcessFunction
(custom fraud rules)
│
▼
Alert Sink
Key Takeaway
Flink's DataStream API is the low-level, imperative interface where you process events one at a time and have full explicit control over keyed state, timers, windows, and custom operator logic. With a KeyedProcessFunction you get state handles like ValueState and ListState and you register and fire timers yourself, so you can implement arbitrary logic such as custom windows, state machines, and pattern detection. The strength is that anything Flink can do you can express here; the cost is verbosity and responsibility.
The Table API and Flink SQL sit on top and are declarative. You express computations as relational queries over tables, and Flink's cost-based optimizer plans and executes them and manages state for stateful operations like joins and aggregations automatically. A windowed aggregation that counts fraud alerts per account per five-minute window is a short SQL statement, whereas the DataStream equivalent needs a keyed window with an aggregate function. The APIs unify batch and streaming, so the same query runs over a bounded table as batch or an unbounded stream as a continuous query.
The two are fully interoperable, which is the practical key. You can write the relational parts in SQL, convert a table to a DataStream when you need custom low-level logic, process it with a process function, and convert back to a table to continue in SQL. At the National Bank of India we used Flink SQL for the windowed aggregation of total transaction amount per account, which the analysts could read and maintain, and the DataStream API for a bespoke fraud rule that needed explicit state and timers.
My default is to write the bulk of a pipeline in Flink SQL for conciseness and optimizer benefits, and drop into the DataStream API only for the specific operators whose logic the declarative layer cannot express cleanly.
Remember one sentence: Use Flink SQL for declarative relational queries and the DataStream API for custom stateful timer logic, converting between them freely within one job.
---