DataStream API 概览
Flink DataStream API 是 Flink 的底层流处理 API，提供对数据流的细粒度控制。对比 SQL/Table API：
| API | 抽象层级 | 适用场景 |
|---|---|---|
| DataStream | 底层（Java/Scala 代码） | 复杂逻辑、自定义状态、精准控制 |
| Table API/SQL | 高层（声明式） | 简单 ETL、分析查询 |
eventd v3 使用 DataStream API，因为需要自定义窗口逻辑和 CEP 模式匹配。
StreamExecutionEnvironment
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 并行度（本地环境默认 = CPU 核数；集群默认看 parallelism.default，通常为 1）
env.setParallelism(4);
// 容错：Checkpoint 每 30 秒（默认 EXACTLY_ONCE）
env.enableCheckpointing(30_000);
时间语义不在 StreamExecutionEnvironment 上设置：Flink 1.12 起每个 Source 用自己的 WatermarkStrategy 决定事件时间，setStreamTimeCharacteristic 已废弃且不改变窗口语义。
getExecutionEnvironment() 自动检测运行环境：
- 本地 IDE → 创建
LocalStreamEnvironment，同 JVM 多线程执行 - 提交到 Flink 集群 → 连接集群 JobManager
本地环境不是 MiniCluster：要验证 Slot 分配、作业提交或 Checkpoint 行为，需要在测试里显式启动 MiniCluster（见 8.3 ）。
Source：数据源
Kafka Source
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
KafkaSource<Event> source = KafkaSource.<Event>builder()
.setBootstrapServers("kafka:9092")
.setTopics("eventd-cleaned")
.setGroupId("eventd-analytics")
.setStartingOffsets(OffsetsInitializer.earliest()) // 或 latest()
.setValueOnlyDeserializer(new EventDeserializationSchema())
.build();
DataStream<Event> stream = env.fromSource(
source,
WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(10))
.withTimestampAssigner((event, ts) -> event.getEventTime()),
"Kafka Source"
);
withTimestampAssigner 不能省：只给 forBoundedOutOfOrderness 而不分配时间戳，Flink 不会把事件时间切成处理时间，而是沿用 Source 已写入记录的时间戳（KafkaSource 用 Kafka record timestamp；集合源常见为 0）。事件时间窗口仍按这个时间戳推进，只是它未必符合业务语义：集合源的时间戳为 0 时窗口可能永远不触发，Kafka 里 timestamp 类型不对时窗口又会落在错误的区间。只有处理时间窗口才按处理时间触发。
Socket Source（测试用）
DataStream<String> text = env.socketTextStream("localhost", 9999);
Collection Source（测试用）
List<Event> events = List.of(
new Event("1", "order", 1000, Instant.now().toEpochMilli()),
new Event("2", "order", 2000, Instant.now().toEpochMilli())
);
DataStream<Event> stream = env.fromCollection(events);
自定义 Source
public class EventSource implements SourceFunction<Event> {
private volatile boolean running = true;
@Override
public void run(SourceContext<Event> ctx) throws Exception {
while (running) {
ctx.collect(generateEvent());
Thread.sleep(100); // 每 100ms 生成一条
}
}
@Override
public void cancel() {
running = false;
}
}
SourceFunction 在 1.20 已废弃（连同 env.addSource）。新代码应实现 Source 接口并用 env.fromSource(source, watermarkStrategy, name)；它支持并行、分区与 Offset 状态。SourceFunction 只在维护旧代码或写最小示例时使用。
Transformation：数据转换
map / filter / flatMap
DataStream<EnrichedEvent> enriched = stream
.filter(event -> event.getAmount() > 0) // 过滤
.map(event -> new EnrichedEvent(event, lookupUser(event.getUserId()))) // 转换
.flatMap((FlatMapFunction<EnrichedEvent, Alert>) (value, out) -> {
if (value.event().getAmount() > 10000) {
out.collect(new Alert("LARGE_ORDER", value));
}
}); // 1 对 N
这里 EnrichedEvent 用 record 定义（record EnrichedEvent(Event event, User user)），它自己的访问器只有 event() 和 user()；金额仍从 Event 的 getAmount() 取。Event 是带 getter 的 POJO，两种风格不要混在同一个类上。
keyBy
DataStream<Event> keyed = stream.keyBy(Event::getUserId);
keyBy 按 key 的 hash 值分区，相同 key 的数据发送到同一分区。后续聚合操作（window、reduce）必须紧跟 keyBy。
reduce / aggregate
keyedStream
.window(TumblingEventTimeWindows.of(Duration.ofMinutes(1)))
.reduce((e1, e2) -> new Event(
e1.getEventId(), e1.getType(),
e1.getAmount() + e2.getAmount(), e1.getEventTime()));
Sink：数据输出
Kafka Sink
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
KafkaSink<String> sink = KafkaSink.<String>builder()
.setBootstrapServers("kafka:9092")
.setRecordSerializer(KafkaRecordSerializationSchema.builder()
.setTopic("eventd-metrics")
.setValueSerializationSchema(new SimpleStringSchema())
.build())
.setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
// EXACTLY_ONCE 必须配事务前缀，否则 build() 直接抛异常
.setTransactionalIdPrefix("eventd-analytics")
.build();
// Sink 的元素类型是 String，先用 map 把 Event 序列化成 JSON
stream.map(JsonMapper::toJson).sinkTo(sink);
打印 / 文件 / Redis Sink
stream.print(); // 打印到控制台（仅测试）
stream.writeAsText("/tmp/output.txt", WriteMode.OVERWRITE); // 文件，写入方式见下方说明
// Redis：Apache Flink 没有官方 Redis 连接器，
// 也没有 RedisSink / RedisMapper 这两个类
// 自定义 Sink 的泛型必须与流元素类型一致（这里是 DataStream<Metric>）
metrics.addSink(new RedisMetricSink()); // 自己实现的 RichSinkFunction，见 7.2
两点容易踩坑：
stream.writeAsText(...)仍然可用，但DataStream.write*这一组方法自 1.19 起已废弃，新代码用FileSink或sinkTo(...)。- 不要照抄旧教程里的
RedisSink/RedisMapper。它们来自 Apache Bahir 的 Redis 连接器，该项目早已停止更新，不支持 Flink 1.20；org.apache.flink组下也没有官方 Redis 连接器。自定义写法见 7.2 侧输出连接器 。
类型系统
Flink 需要知道数据的类型信息（TypeInformation）来选择序列化器和优化算子：
// 自动推断（Java API）
DataStream<Event> stream = env.fromCollection(events); // 自动推断 Event 类型
// 显式声明（POJO 更精确）
env.getConfig().registerTypeWithKryoSerializer(Event.class, EventSerializer.class);
POJO 规则：
- 类是 public 的
- 有无参构造函数
- 所有字段是 public 或有 getter/setter
- 字段类型也是 POJO 或基本类型
public class Event {
public String eventId;
public String type;
public String userId;
public String symbol;
public double amount;
public long eventTime;
public Event() {} // 必须有：Flink 反序列化时要能实例化
public Event(String eventId, String type, double amount, long eventTime) {
this.eventId = eventId;
this.type = type;
this.amount = amount;
this.eventTime = eventTime;
}
public String getEventId() { return eventId; }
public String getType() { return type; }
public String getUserId() { return userId; }
public String getSymbol() { return symbol; }
public double getAmount() { return amount; }
public long getEventTime() { return eventTime; }
}
Event 字段类型都是 String/double/long，Flink 会把它当 POJO 用 PojoSerializer，比交给 Kryo 更省空间也更快。registerTypeWithKryoSerializer 只在这类自动推断失效（字段是接口、泛型擦除）时才需要。
并行度与 Slot
并行度 = 算子实例数（同时处理的分区数）
Slot = TaskManager 上的资源槽（CPU + 内存）
env.setParallelism(4):
Source(4) → Map(4) → KeyBy(4) → Sink(4)
↑ ↑ ↑ ↑
4 个实例 4 个实例 4 个实例 4 个实例
setParallelism 设的是默认值，不是只作用于某一个算子：没有单独覆盖的算子都会用 4。Source 的并行度还受分区数约束——Kafka Topic 只有 2 个分区时，即使默认并行度是 4，Source 实际也只会用到 2 个 subtask，多出来的处于空闲状态。
单个算子可覆盖并行度：
stream.map(new ExpensiveMapper()).setParallelism(8); // 只给 map 配 8 并行
执行
// 阻塞执行（推荐 main 方法）
env.execute("eventd-analytics-job");
// 异步执行（测试或 Web 应用）
JobClient client = env.executeAsync("eventd-analytics-job");
client.getJobExecutionResult().get(); // 等待完成
本地环境测试
@Test
void testWordCount() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
DataStream<String> text = env.fromElements("hello world", "hello flink");
DataStream<Tuple2<String, Integer>> counts = text
.flatMap((FlatMapFunction<String, Tuple2<String, Integer>>) (value, out) -> {
for (String word : value.split(" ")) {
out.collect(Tuple2.of(word, 1));
}
})
.returns(Types.TUPLE(Types.STRING, Types.INT))
.keyBy(v -> v.f0)
.sum(1);
// 用 CollectSink 收集结果（实现见 8.3；集合必须是静态的，
// 因为算子函数会被序列化后在新对象里执行）
CollectSink.VALUES.clear();
counts.addSink(new CollectSink<>());
env.execute("test");
List<Tuple2<String, Integer>> result = CollectSink.values();
assertTrue(result.contains(Tuple2.of("hello", 2)));
assertTrue(result.contains(Tuple2.of("world", 1)));
}
小结
| 概念 | API | 说明 |
|---|---|---|
| Source | env.fromSource() / env.socketTextStream() | 数据输入 |
| Transformation | map / filter / flatMap / keyBy | 数据转换 |
| Sink | stream.sinkTo() / stream.print() | 数据输出 |
| 并行度 | env.setParallelism() | 全局或单算子 |
| POJO | public 类 + 无参构造 + getter | 类型推断 |
| 执行 | env.execute() / executeAsync() | 阻塞/异步 |
下一节 4.2 Flink 时间语义与 Watermark 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。