Definition
Flink provides a layered set of APIs for different levels of control.
- DataSet API (deprecated) – The original batch-only API for finite collections with operators like map, filter, groupBy, and join, with no event-time or watermark support.
- DataStream API (recommended) – The unified low-level API for both unbounded streaming and bounded batch, supporting event-time, watermarks, keyed state, timers, windows, and ProcessFunction.
- Table API – A fluent, language-integrated relational API in Java, Scala, or Python that compiles to the same runtime.
- Flink SQL – Standard SQL over dynamic tables, compiled to the same runtime as Table API.
The DataStream API is for fine-grained control; the Table API and Flink SQL are for declarative analytics.
At the National Bank of India, the old DataSet API was like a machine built only for processing a fixed pile of yesterday's transactions overnight: a finished stack with a clear end. The DataStream API is a newer machine that handles both a never-ending live feed and a fixed pile, using the same controls, and it lets an engineer wire up every detail by hand. The Table API and Flink SQL sit on top as a friendly control panel where an analyst just types the query they want, like total per branch, and the machine figures out the wiring.
Understanding the Concepts
What is the DataSet API?
The DataSet API was Flink's original low-level API for bounded batch processing. It modeled data as finite collections transformed by operators like map, flatMap, filter, groupBy, reduce, and join. The runtime was free to use batch-optimized execution like sort-merge joins and full-blocking shuffles because it knew the data had a definite end.
// Old DataSet API (deprecated) for batch processing
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.DataSet;
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
DataSet<Transaction> txns = env.readCsvFile("/data/bank/transactions.csv")
.pojoType(Transaction.class, "accountId", "amount", "txnDate");
DataSet<Tuple2<String, Double>> totals = txns
.filter(t -> t.amount.compareTo(BigDecimal.ZERO) > 0)
.groupBy("accountId")
.sum("amount");
totals.print();
The DataSet API had no concept of event time, watermarks, or continuous state because batch data does not stream. Flink has deprecated it because the DataStream API now covers both streaming and batch.
What is the DataStream API?
The DataStream API is the recommended low-level API that unifies unbounded streaming and bounded batch in one programming model. It provides the full streaming toolkit: event-time and processing-time semantics, watermarks, keyed state, timers, windows, and ProcessFunction.
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.api.common.RuntimeExecutionMode;
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// Unified: set BATCH mode for bounded data
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
DataStream<Transaction> txns = env.readFile(
new TransactionInputFormat(), "/data/bank/transactions.csv");
DataStream<AccountTotal> totals = txns
.keyBy(Transaction::getAccountId)
.sum("amount");
totals.print();
env.execute("National Bank Account Totals");
How do you use keyed state and timers with DataStream?
The DataStream API gives fine-grained control over per-key state and timers through ProcessFunction. This is essential for use cases like fraud detection at the National Bank of India, where you need custom state machines per account.
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.util.Collector;
import java.math.BigDecimal;
public class FraudDetector extends KeyedProcessFunction<String, Transaction, Alert> {
private ValueState<BigDecimal> runningTotal;
private ValueState<Long> timerTimestamp;
@Override
public void open(Configuration params) {
runningTotal = getRuntimeContext().getState(
new ValueStateDescriptor<>("total", BigDecimal.class));
timerTimestamp = getRuntimeContext().getState(
new ValueStateDescriptor<>("timer", Long.class));
}
@Override
public void processElement(Transaction txn, Context ctx, Collector<Alert> out) throws Exception {
BigDecimal current = runningTotal.value() != null ? runningTotal.value() : BigDecimal.ZERO;
BigDecimal updated = current.add(txn.getAmount());
runningTotal.update(updated);
// Flag if daily total exceeds threshold for this account
if (updated.compareTo(new BigDecimal("5000000")) > 0) {
out.collect(new Alert(txn.getAccountId(), "HIGH_DAILY_TOTAL", updated));
}
// Set a timer to reset the daily total at midnight
long midnight = nextMidnight(ctx.timerService().currentProcessingTime());
ctx.timerService().registerProcessingTimeTimer(midnight);
timerTimestamp.update(midnight);
}
@Override
public void onTimer(long timestamp, OnTimerContext ctx, Collector<Alert> out) {
runningTotal.clear();
}
}
What are Table API and Flink SQL?
The Table API and Flink SQL are the declarative layer built on top. They operate over dynamic tables and compile down to the same underlying runtime as the DataStream API. You choose them when the computation is relational.
# Flink SQL example for banking analytics
from pyflink.table import EnvironmentSettings, TableEnvironment
env_settings = EnvironmentSettings.in_streaming_mode()
table_env = TableEnvironment.create(env_settings)
table_env.execute_sql("""
CREATE TABLE transactions (
account_id STRING,
amount DECIMAL(18,2),
branch_code STRING,
txn_time TIMESTAMP(3),
WATERMARK FOR txn_time AS txn_time - INTERVAL '5' SECOND
) WITH (
'connector' = 'kafka',
'topic' = 'bank-transactions',
'format' = 'json'
)
""")
# Declarative SQL: the optimizer handles the plan
result = table_env.execute_sql("""
SELECT branch_code,
TUMBLE_START(txn_time, INTERVAL '1' MINUTE) AS window_start,
SUM(amount) AS total_amount,
COUNT(*) AS txn_count
FROM transactions
GROUP BY branch_code, TUMBLE(txn_time, INTERVAL '1' MINUTE)
""")
When do you choose DataStream over Table API?
You drop to the DataStream API when you need something SQL cannot express cleanly: custom per-key state machines, data-dependent timers, bespoke windowing or triggering, complex event logic beyond MATCH_RECOGNIZE, or precise control over operator behavior.
| Criterion | Table API / SQL | DataStream API |
|---|---|---|
| Relational queries | Best fit | Overkill |
| Custom state machines | Not possible | Full control |
| Data-dependent timers | Not possible | Full control |
| Optimizer benefits | Yes | No |
| Code conciseness | Very concise | Verbose |
| Bespoke windowing | Limited | Full flexibility |
How do you mix both APIs in one job?
Flink lets you convert between a Table and a DataStream, so you can do the heavy relational work in SQL and switch to a ProcessFunction for custom logic.
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
// SQL for relational aggregation
Table branchTotals = tableEnv.sqlQuery(
"SELECT branch_code, SUM(amount) AS total FROM transactions GROUP BY branch_code");
// Convert to DataStream for custom fraud logic
DataStream<Row> stream = tableEnv.toDataStream(branchTotals);
DataStream<Alert> alerts = stream
.keyBy(row -> row.getFieldAs("branch_code"))
.process(new BranchFraudDetector());
alerts.print();
env.execute("Bank Mixed Pipeline");
What about batch mode on DataStream?
The DataStream API has a batch execution mode: when the sources are bounded, you set RuntimeExecutionMode.BATCH and Flink executes the same program with batch optimizations like sort-merge joins and blocking shuffles.
// Same DataStream code, different execution mode
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.BATCH); // batch optimizations
// The same pipeline code works for both streaming and batch
DataStream<Transaction> txns = env.readFile(
new TransactionInputFormat(), "/data/bank/historical_transactions.csv");
txns.keyBy(Transaction::getAccountId)
.sum("amount")
.print();
env.execute("Batch Bank Processing");
What should a senior engineer know?
- For new projects, do not use the DataSet API at all: use the DataStream API in batch mode for bounded jobs.
- Always prefer the highest-level API that fits: declarative code is cheaper to maintain and benefits from optimization.
- Batch execution mode changes scheduling and state behavior, so understand that bounded mode may use different shuffle and failover strategies.
- You can interleave the APIs within one job, using SQL for analytics and ProcessFunction for custom logic.
- The Table API and SQL share the same planner and produce the same execution plan.
// Verify you are NOT using the deprecated DataSet API
// Correct: StreamExecutionEnvironment (unified)
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// Wrong: ExecutionEnvironment (deprecated DataSet API)
// ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
Visual Overview
┌──────────────────────────────────────────────┐ │ Flink API Layers │ │ │ │ ┌─────────────────────────────────────┐ │ │ │ Flink SQL / Table API │ │ │ │ (declarative, SQL-like) │ │ │ │ SELECT, JOIN, GROUP BY, WINDOW │ │ │ └──────────────┬──────────────────────┘ │ │ │ compiles to │ │ ┌──────────────▼──────────────────────┐ │ │ │ DataStream API │ │ │ │ (unified low-level) │ │ │ │ State, Timers, ProcessFunction │ │ │ │ Streaming + Batch mode │ │ │ └──────────────┬──────────────────────┘ │ │ │ │ │ ┌──────────────▼──────────────────────┐ │ │ │ Flink Runtime │ │ │ │ Task scheduling, checkpointing │ │ │ │ State backends, network shuffle │ │ │ └─────────────────────────────────────┘ │ │ │ │ ┌─────────────────────────────────────┐ │ │ │ DataSet API (DEPRECATED) │ │ │ │ Batch only, no event-time │ │ │ └─────────────────────────────────────┘ │ └──────────────────────────────────────────────┘
Key Takeaway
Flink has a layered set of APIs, and the modern story is a unified one. The DataSet API was the original low-level API for bounded batch processing with operators like map, groupBy, and join, but it had no concept of event time, watermarks, or continuous state, and it is now deprecated. The DataStream API replaces it as the recommended low-level API, unifying unbounded streaming and bounded batch in one programming model, with event-time semantics, watermarks, keyed state, timers, windows, and ProcessFunction.
Above the DataStream API sit the Table API and Flink SQL, the declarative layer that compiles down to the same runtime. The choice between layers is about how much control you need. Use the Table API or Flink SQL whenever the computation is relational: filters, projections, joins, aggregations, and windows, because the declarative form is far more concise, easier to maintain, and the optimizer can improve the plan. Drop to the DataStream API only when you need something SQL cannot express cleanly: a custom per-key state machine, data-dependent timers, or bespoke windowing.
At the National Bank of India this distinction was a daily decision. For straightforward analytics like total transaction amount per branch, we used one Flink SQL GROUP BY statement. But for a fraud rule that flagged transactions exceeding a dynamic per-account threshold, SQL could not express the evolving per-key state, so we used the DataStream API with a keyed ProcessFunction. We kept the relational aggregation in SQL and the custom rule in a ProcessFunction within the same pipeline, which is exactly the recommended mix.
The senior points to stress are: never use the DataSet API for new projects, always prefer the highest-level API that fits, and know that you can interleave SQL and DataStream in one job.
Remember one sentence: the DataSet API is deprecated, the DataStream API unifies batch and streaming with full low-level control, and the Table API and Flink SQL provide the declarative layer you should default to unless you need custom state or timers.
---