企业官网建设流程全解析

一、背景

Flink 提供了四层 API 抽象，自底向上分别为：

┌─────────────────────────────────────────────────────────┐ │ SQL（最高层声明式） │ ├─────────────────────────────────────────────────────────┤ │ Table API（关系型操作） │ ├─────────────────────────────────────────────────────────┤ │ DataStream API（流处理核心 API） ← 本文重点 │ ├─────────────────────────────────────────────────────────┤ │ ProcessFunction（最底层、最灵活） │ └─────────────────────────────────────────────────────────┘

DataStream API 的定位：在灵活性与易用性之间取得平衡——比 SQL/Table API 更底层、更灵活，可直接操作事件时间、状态和定时器；比 ProcessFunction 更具结构化，提供了丰富的预定义算子。

尽管 Flink 社区在大力推进 SQL/Table API 的功能覆盖，但在以下场景中 DataStream API 仍不可替代：

• 复杂的自定义序列化/反序列化逻辑
• 需要直接操作底层 State 后端（如 MapState 的 TTL 精细控制）
• 异构多流的非等值关联（超出 SQL JOIN 能力范围）
• 对延迟有极致要求的低延迟管道（需要精细控制 buffer timeout）

二、核心原理与架构

DataStream 执行架构图：

核心概念：

Flink DataStream 程序采用懒执行机制：调用 map()、filter() 等方法时，并不会立即执行计算，而是构建一个 DAG（有向无环图）。只有在显式调用 env.execute() 时，才会将整个 DAG 提交给集群执行。

这种设计的优势：

• 允许 Flink 对整个 DAG 做全局优化（算子链接、资源分配）
• 支持在提交前进行计划验证

三、程序骨架与执行模型

每个 Flink DataStream 程序遵循以下五步模式：

// 1. 获取执行环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 2. 配置执行参数 env.setParallelism(4); env.enableCheckpointing(60000); // 60s 一次 checkpoint // 3. 读取数据源 DataStream<String> source = env.addSource(new FlinkKafkaConsumer<>(...)); // 4. 定义转换逻辑 DataStream<Result> result = source .map(...) .keyBy(...) .window(...) .reduce(...); // 5. 输出结果 & 触发执行 result.addSink(new FlinkKafkaProducer<>(...)); env.execute("My Flink Job");

执行流程图：

四、数据源 Source API

从 Flink 1.12 起推荐使用新的 Source 接口（统一批/流，支持更好的扩展性），旧的 SourceFunction 已被标记为过时。新接口将 Source 拆分为 SplitEnumerator（分片枚举）和 SourceReader（数据读取）两个组件。

┌──────────────────────────────────────────────────────┐ │ Source (FLIP-27) │ ├──────────────────────┬───────────────────────────────┤ │ SplitEnumerator │ SourceReader │ │ (运行在 JobManager) │ (运行在 TaskManager) │ │ │ │ │ • 发现/分配分片 │ • 从分配的分片读取数据 │ │ • 协调多个 Reader │ • 维护分片级别的状态 │ │ • 支持动态发现新分片│ • 发送数据到下游算子 │ └──────────────────────┴───────────────────────────────┘

常用预定义 Source：

// Kafka Source（推荐使用新接口） KafkaSource<String> kafkaSource = KafkaSource.<String>builder() .setBootstrapServers("broker1:9092") .setTopics("input-topic") .setGroupId("my-group") .setStartingOffsets(OffsetsInitializer.earliest()) .setValueOnlyDeserializer(new SimpleStringSchema()) .build(); DataStream<String> stream = env.fromSource( kafkaSource, WatermarkStrategy.noWatermarks(), "Kafka Source" ); // 文件 Source FileSource<String> fileSource = FileSource .forRecordStreamFormat(new TextLineInputFormat(), new Path("/data/input")) .monitorContinuously(Duration.ofSeconds(10)) .build(); // Collection Source（用于测试） DataStream<Integer> testStream = env.fromElements(1, 2, 3, 4, 5);

自定义Source示例：

// 基于新 Source 接口（简化示意） public class MySource implements Source<MyEvent, MySplit, MyEnumState> { @Override public Boundedness getBoundedness() { return Boundedness.CONTINUOUS_UNBOUNDED; // 无界流 } @Override public SplitEnumerator<MySplit, MyEnumState> createEnumerator( SplitEnumeratorContext<MySplit> context) { return new MySplitEnumerator(context); } @Override public SourceReader<MyEvent, MySplit> createReader( SourceReaderContext context) { return new MySourceReader(context); } // ... 其他方法 }

五、转换算子 Transformation

Filter：条件过滤

// 过滤无效订单 DataStream<Order> validOrders = orders.filter(order -> order.getAmount() > 0 && order.getUserId() != null );

Map：一对一转换（适用场景：数据清洗、格式转换、字段提取等一对一映射操作。）

// 将JSON字符串解析为POJO DataStream<Order> orders = rawStream.map(new MapFunction<String, Order>() { @Override public Order map(String json) throws Exception { return objectMapper.readValue(json, Order.class); } }); // Lambda 简写 DataStream<Integer> doubled = intStream.map(value -> value * 2);

FlatMap：一对多转换（适用场景：一条记录展开为多条（如日志解析、订单拆分）、条件过滤+转换合一。）

// 将订单拆分为多个订单项事件 DataStream<OrderItem> items = orders.flatMap( new FlatMapFunction<Order, OrderItem>() { @Override public void flatMap(Order order, Collector<OrderItem> out) { for (OrderItem item : order.getItems()) { out.collect(item); } } } );

KeyBy：逻辑分区（key 类型不能是数组；POJO 作为 key 时必须正确重写 hashCode() 方法；KeyBy 会触发网络 Shuffle（hash 分区））

// 按用户ID分区 KeyedStream<Order, String> keyedOrders = orders.keyBy(Order::getUserId); // 使用元组字段 KeyedStream<Tuple2<String, Integer>, String> keyed = tupleStream.keyBy(t -> t.f0);

Operator Chaining：Flink 默认会将可以链接的算子合并在同一个 Task 线程中执行，以减少线程切换和序列化开销。

链接前：Source → Map → Filter → KeyBy → Window → Sink ↓ (network shuffle) 链接后：[Source → Map → Filter] → [KeyBy → Window → Sink] (一个 Operator Chain) (另一个 Operator Chain)

可通过以下方式控制链接行为：

// 全局禁用算子链 env.disableOperatorChaining(); // 单个算子断开前向链 stream.map(...).startNewChain(); // 单个算子完全禁用链接 stream.map(...).disableChaining();

六、数据汇 Sink API

与 Source 类似，Sink 也经历了接口重构。从1.14版本开始使用新的Sink接口（新统一接口，支持精确一次语义），旧的SinkFunction接口被标记为过时。

常用 Sink 示例：

// Kafka Sink（新接口） KafkaSink<String> kafkaSink = KafkaSink.<String>builder() .setBootstrapServers("broker1:9092") .setRecordSerializer( KafkaRecordSerializationSchema.builder() .setTopic("output-topic") .setValueSerializationSchema(new SimpleStringSchema()) .build() ) .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE) // 精确一次 .setTransactionalIdPrefix("my-app") .build(); stream.sinkTo(kafkaSink); // FileSink（精确一次 + 文件滚动） FileSink<String> fileSink = FileSink .forRowFormat(new Path("/output"), new SimpleStringEncoder<String>("UTF-8")) .withRollingPolicy( DefaultRollingPolicy.builder() .withRolloverInterval(Duration.ofMinutes(15)) .withInactivityInterval(Duration.ofMinutes(5)) .withMaxPartSize(MemorySize.ofMebiBytes(1024)) .build() ) .build(); stream.sinkTo(fileSink);

writeAsText() / print() 等便捷方法不参与 Checkpoint，仅适合调试，不保证 Exactly-Once。生产环境务必使用 sinkTo() 或 addSink() 结合支持事务的 Connector。