java

附录 A：Flink DataStream API 速查

Flink DataStream API 常用操作速查卡：Environment 创建、Source 与 Sink、Transformation、Window、State、Watermark 策略、Checkpoint 配置及 Metrics 注册的代码模板与参数速查。

Leeting Yan 2026-11-01 2 分钟阅读 811 字

Environment

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(4);
env.enableCheckpointing(30_000);          // 默认就是 EXACTLY_ONCE
env.setStateBackend(new EmbeddedRocksDBStateBackend());
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Duration.ofSeconds(10)));
// 时间语义不在 Environment 上设置：由每个 Source 的 WatermarkStrategy 决定

方法	默认	作用
`getExecutionEnvironment()`	-	自动检测环境
`createLocalEnvironment(n)`	-	本地 n 并行度
`setParallelism(n)`	本地 = CPU 核数；集群 = `parallelism.default`（默认 1）	全局并行度
`enableCheckpointing(ms)`	无	开启 Checkpoint
`setStateBackend(backend)`	HashMap	状态后端
`WatermarkStrategy`（Source 侧）	无	事件时间与 Watermark 策略；没有全局时间语义开关

速查卡里的 Event 以 9.1 的 v3 record 为准，访问器是 amount()/eventTime()/userId()。第 4~8 章用的是同名 POJO，对应写法是 getAmount()/getEventTime()/getUserId()，两种不要混用。

Source

// Kafka Source
KafkaSource<Event> kafkaSource = KafkaSource.<Event>builder()
    .setBootstrapServers("kafka:9092")
    .setTopics("topic")
    .setGroupId("group")
    .setStartingOffsets(OffsetsInitializer.earliest())
    .setValueOnlyDeserializer(new EventDeserializationSchema())
    .build();

DataStream<Event> stream = env.fromSource(kafkaSource, watermarkStrategy, "kafka");

// Collection Source
DataStream<Event> stream = env.fromCollection(list);

// Socket Source
DataStream<String> stream = env.socketTextStream("localhost", 9999);

Transformation

stream
    .map(e -> transform(e))              // 1:1 转换
    .filter(e -> e.amount() > 0)         // 过滤
    .flatMap((FlatMapFunction<T, R>) ...) // 1:N
    .keyBy(Event::userId)                 // 分区
    .window(TumblingEventTimeWindows.of(Duration.ofMinutes(1)))  // 窗口
    .aggregate(new MyAggregate())         // 聚合
    .addSink(sink);                        // 输出

操作	输入	输出	状态
`map`	T	R	无
`filter`	T	T	无
`flatMap`	T	R	无
`keyBy`	T	KeyedStream	无
`reduce`	KeyedStream	T	有
`aggregate`	KeyedStream	R	有
`process`	KeyedStream	R	有（自定义）
`window`	KeyedStream	WindowedStream	有

Window

// 滚动窗口
stream.keyBy(...).window(TumblingEventTimeWindows.of(Duration.ofMinutes(1)));

// 滑动窗口
stream.keyBy(...).window(SlidingEventTimeWindows.of(Duration.ofMinutes(5), Duration.ofMinutes(1)));

// 会话窗口
stream.keyBy(...).window(EventTimeSessionWindows.withGap(Duration.ofMinutes(30)));

// 全局窗口 + 触发器
stream.keyBy(...).window(GlobalWindows.create()).trigger(CountTrigger.of(100));

窗口	API	适用
滚动	`TumblingEventTimeWindows`	固定周期统计
滑动	`SlidingEventTimeWindows`	移动平均
会话	`EventTimeSessionWindows`	用户行为
全局	`GlobalWindows`	自定义触发

Watermark

// 有界乱序
WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(10))
    .withIdleness(Duration.ofMinutes(5))
    .withTimestampAssigner((event, ts) -> event.eventTime());

// 单调递增
WatermarkStrategy.<Event>forMonotonousTimestamps()
    .withTimestampAssigner((event, ts) -> event.eventTime());

策略	适用
`forBoundedOutOfOrderness`	有乱序
`forMonotonousTimestamps`	严格递增
`noWatermarks`	Processing Time

State

// ValueState
ValueStateDescriptor<Long> descriptor = new ValueStateDescriptor<>("count", Types.LONG);
ValueState<Long> state = getRuntimeContext().getState(descriptor);

// ListState
ListStateDescriptor<Event> descriptor = new ListStateDescriptor<>("events", Event.class);
ListState<Event> state = getRuntimeContext().getListState(descriptor);

// MapState
MapStateDescriptor<String, Rule> descriptor = new MapStateDescriptor<>("rules", String.class, Rule.class);
MapState<String, Rule> state = getRuntimeContext().getMapState(descriptor);

状态类型	结构	接口
ValueState	key → value	`value()`, `update(v)`
ListState	key → List	`add(v)`, `get()`, `update(list)`
MapState	key → Map	`put(k,v)`, `get(k)`, `entries()`
ReducingState	key → reduce 结果	`add(v)`

Checkpoint

env.enableCheckpointing(30000);
env.getCheckpointConfig().setCheckpointingConsistencyMode(
    org.apache.flink.core.execution.CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setCheckpointTimeout(60000);
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(10000);
env.getCheckpointConfig().enableUnalignedCheckpoints();
env.getCheckpointConfig().setExternalizedCheckpointCleanup(
    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

配置	参数	作用
`enableCheckpointing`	ms	开启 + 间隔（默认 EXACTLY_ONCE）
`setCheckpointingConsistencyMode`	EXACTLY_ONCE / AT_LEAST_ONCE	语义；旧的 `setCheckpointingMode(streaming.api.CheckpointingMode)` 已废弃
`setCheckpointTimeout`	ms	单次超时，超时即取消并记一次失败
`enableUnalignedCheckpoints`	-	降低反压影响；1.20 默认 `aligned-checkpoint-timeout=0`，开启后每次都直接走非对齐

Metrics

Counter counter = getRuntimeContext().getMetricGroup().counter("name");
Gauge<Double> gauge = getRuntimeContext().getMetricGroup().gauge("name", () -> compute());

Sink

// Kafka Sink
KafkaSink<Event> sink = KafkaSink.<Event>builder()
    .setBootstrapServers("kafka:9092")
    .setRecordSerializer(...)
    .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
    .setTransactionalIdPrefix("eventd-analytics")  // EXACTLY_ONCE 必填
    .build();
stream.sinkTo(sink);

// 打印
stream.print();

// 自定义 Sink（SinkFunction 在 1.20 已废弃，新代码优先 Sink V2）
stream.addSink(new RichSinkFunction<Event>() {
    @Override
    public void open(OpenContext ctx) { /* 初始化连接 */ }

    @Override
    public void invoke(Event value, Context ctx) { /* 写入 */ }
});

执行

env.execute("job name");        // 阻塞
env.executeAsync("job name");   // 非阻塞

返回目录。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章返回首页