4. Flink DataStream API 基础：编程模型与核心组件

DataStream API 是 Flink 实时计算的核心抽象，它提供了一套声明式的编程接口，用于定义无界或有界数据流的处理逻辑。本章将深入剖析 DataStream 编程模型，并详细讲解 Source、Transformation 和 Sink 三大核心组件的原理与实战用法。

4.1 DataStream 编程模型

Flink 的 DataStream 程序本质上是一个数据流转换图，由以下三个基本部分构成：

• Source（数据源）：负责从外部系统读取数据，并将其转换为 DataStream。
• Transformation（转换算子）：对 DataStream 进行各种运算（如 map、filter、keyBy、window），生成新的 DataStream。
• Sink（数据汇）：将最终结果输出到外部系统（如 Kafka、数据库、文件）。

下图展示了 DataStream 程序的典型执行流程：

4.2 Source：数据源接入

Source 是 DataStream 的起点。Flink 提供了丰富的内置 Source，同时也支持自定义 Source。根据数据消费方式，Source 可分为有界和无界两类。

4.2.1 内置 Source 对比

Source 类型	使用场景	并行度	有界/无界	常用 API
Kafka Source	高吞吐消息队列消费	可配置（建议与分区数一致）	无界	KafkaSource.builder()
Socket Source	测试、调试	1（仅支持单并行）	无界	env.socketTextStream()
File Source	批量读取文件、历史数据回放	可配置	有界	FileSource.forRecordStreamFormat()
Collection Source	单元测试、本地开发	1	有界	env.fromCollection()

4.2.2 Kafka Source 实战示例

在实时额度计算场景中，Kafka 是最常用的 Source。以下代码展示了如何构建一个 Kafka Source：

// 1. 创建执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 2. 构建 Kafka Source
KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
    .setBootstrapServers("localhost:9092")
    .setTopics("order-topic")
    .setGroupId("flink-credit-group")
    .setStartingOffsets(OffsetsInitializer.latest())
    .setValueOnlyDeserializer(new SimpleStringSchema())
    .build();

// 3. 接入数据流
DataStream<String> orderStream = env.fromSource(
    kafkaSource,
    WatermarkStrategy.noWatermarks(),
    "Kafka Order Source"
);

4.3 Transformation：数据转换与处理

Transformation 是 DataStream API 的核心，它定义了数据处理的业务逻辑。Flink 提供了丰富的转换算子，按功能可分为单流转换、多流转换和分区转换三大类。

4.3.1 核心算子分类

类别	算子	功能描述	典型应用
单流转换	map	一对一转换，输入一条输出一条	字段提取、格式转换
	filter	根据条件过滤数据	过滤无效订单、风控拦截
	flatMap	一对多转换，一条输入可输出零到多条	拆分复合事件、日志解析
多流转换	union	合并多个同类型流	合并多渠道订单
	connect / coMap	连接两个不同类型流，共享状态	订单流与账户流关联
分区转换	keyBy	按 key 分区，相同 key 进入同一分区	按用户 ID 分组计算额度
	rebalance	轮询分区，均匀分配数据	解决数据倾斜

4.3.2 额度计算中的 Transformation 链

以下代码展示了一个典型的实时额度计算 Transformation 链：

// 假设 orderStream 是来自 Kafka 的 JSON 字符串
DataStream<OrderEvent> orderEventStream = orderStream
    .map(json -> JsonUtils.parse(json, OrderEvent.class))   // 解析 JSON
    .filter(order -> order.getAmount() > 0)                 // 过滤无效订单
    .assignTimestampsAndWatermarks(
        WatermarkStrategy.<OrderEvent>forBoundedOutOfOrderness(Duration.ofSeconds(5))
            .withTimestampAssigner((event, ts) -> event.getEventTime())
    );

// 按用户 ID 分组，计算实时额度
DataStream<CreditUsage> creditStream = orderEventStream
    .keyBy(OrderEvent::getUserId)
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .aggregate(new CreditAggregateFunction())
    .name("credit-window-agg");

4.4 Sink：结果输出

Sink 是 DataStream 的终点，负责将处理结果写入外部系统。Flink 同样提供了丰富的内置 Sink，并支持自定义 Sink（实现 SinkFunction 接口）。

4.4.1 常用 Sink 对比

Sink 类型	输出目标	事务支持	适用场景
Kafka Sink	Kafka Topic	精确一次（两阶段提交）	结果下发给下游系统
JDBC Sink	关系型数据库	至少一次	额度结果持久化
File Sink	HDFS / 本地文件	精确一次	批量报表、数据备份
Print Sink	标准输出	无	调试、测试

4.4.2 Kafka Sink 实战示例

将实时额度计算结果写回 Kafka，供其他服务消费：

// 构建 Kafka Sink
KafkaSink<CreditUsage> kafkaSink = KafkaSink.<CreditUsage>builder()
    .setBootstrapServers("localhost:9092")
    .setRecordSerializer(KafkaRecordSerializationSchema.builder()
        .setTopic("credit-result-topic")
        .setValueSerializationSchema(new CreditUsageSerializationSchema())
        .build()
    )
    .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
    .build();

// 将结果写入 Kafka
creditStream.sinkTo(kafkaSink).name("credit-kafka-sink");

// 触发执行
env.execute("Flink Credit Calculation Job");

4.5 完整作业结构图

下图展示了包含 Source、Transformation 和 Sink 的完整实时额度计算作业结构：

4.6 本章小结

• 编程模型：DataStream 程序由 Source、Transformation、Sink 三部分组成，通过 env.execute() 触发执行。
• Source：Kafka Source 是实时额度计算的首选，支持精确一次语义和 offset 管理。
• Transformation：核心算子包括 map、filter、keyBy、window 和 aggregate，它们构成了额度计算的业务逻辑链。
• Sink：Kafka Sink 和 JDBC Sink 是额度计算结果输出的常用选择，建议开启精确一次保证。