3、Flink编程模型：DataStream API基础、Source/Transformation/Sink三要素、执行环境（StreamExecutionEnvironment）

Flink 的 DataStream API 是构建实时特征计算的核心编程接口。本章将深入剖析 DataStream API 的三大核心要素：Source（数据源）、Transformation（转换算子）和Sink（数据输出），并详细讲解执行环境 StreamExecutionEnvironment 的配置与使用。掌握这些基础是编写高效、可靠的实时特征计算作业的前提。

3.1 DataStream API 编程模型总览

Flink 的 DataStream 程序本质上是一个有向无环图（DAG），数据从 Source 流入，经过一系列 Transformation 处理，最终由 Sink 输出。整个程序的执行由 StreamExecutionEnvironment 统一管理。

3.2 StreamExecutionEnvironment：执行环境

StreamExecutionEnvironment 是所有 Flink DataStream 程序的入口。它负责配置作业的执行参数、设置并行度、注册定时器、管理 checkpoint 以及触发作业执行。

3.2.1 创建执行环境

根据运行模式不同，Flink 提供三种创建方式：

// 推荐方式：自动检测运行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 本地模式：显式指定并行度为 4
StreamExecutionEnvironment localEnv = StreamExecutionEnvironment.createLocalEnvironment(4);

// 远程模式：提交到 Flink 集群
StreamExecutionEnvironment remoteEnv = StreamExecutionEnvironment
    .createRemoteEnvironment("jobmanager-host", 6123, "path/to/job.jar");

3.2.2 核心配置项

执行环境提供丰富的配置接口，以下是在实时特征计算中常用的配置：

3.3 Source：数据源接入

Source 是 Flink 作业的数据入口，负责从外部系统读取数据并转换为 DataStream。Flink 内置了丰富的 Source 实现，同时也支持自定义 Source。

3.3.1 内置 Source 分类

3.3.2 Kafka Source 实战示例

在实时特征计算中，Kafka 是最主流的数据源。以下代码展示了如何配置一个带有水位线（Watermark）的 Kafka Source：

Properties props = new Properties();
props.setProperty("bootstrap.servers", "localhost:9092");
props.setProperty("group.id", "feature-group");

DataStream<String> sourceStream = env
    .addSource(new FlinkKafkaConsumer<>(
        "user-click-topic",
        new SimpleStringSchema(),
        props
    ))
    .assignTimestampsAndWatermarks(
        WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ofSeconds(5))
            .withTimestampAssigner((event, timestamp) -> extractEventTime(event))
    );

// 设置 Source 并行度与 Kafka 分区数保持一致
sourceStream.setParallelism(3);

3.4 Transformation：转换算子

Transformation 是 Flink 作业的核心处理逻辑，负责对数据流进行转换、过滤、聚合、分组等操作。DataStream API 提供丰富的算子库，满足各种特征计算需求。

3.4.1 常用算子分类

3.4.2 典型特征计算链

以下代码展示了一个完整的特征计算 Transformation 链：从原始日志中提取用户行为，按用户分组，计算滑动窗口内的行为计数特征：

DataStream<UserClick> clicks = sourceStream
    .map(line -> parseClickEvent(line))          // 解析为 POJO
    .filter(click -> click.getUserId() != null); // 过滤无效数据

DataStream<UserFeature> features = clicks
    .keyBy(UserClick::getUserId)                 // 按用户分组
    .window(SlidingEventTimeWindows.of(
        Time.minutes(10), Time.minutes(1)))      // 10分钟窗口，1分钟滑动
    .aggregate(new ClickCountAggregate())        // 自定义聚合计算特征
    .map(feature -> enrichFeature(feature));     // 特征富集

3.5 Sink：数据输出

Sink 负责将处理后的特征数据写入外部存储系统，供在线推理服务或离线分析使用。Sink 的稳定性和写入性能直接影响特征计算的端到端延迟。

3.5.1 常用 Sink 对比

3.5.2 Redis Sink 实战示例

实时特征计算中，Redis 是最常用的在线特征存储。以下示例展示了如何将特征写入 Redis：

// 定义 Redis 连接配置
FlinkJedisPoolConfig jedisConfig = new FlinkJedisPoolConfig.Builder()
    .setHost("redis-host")
    .setPort(6379)
    .setDatabase(0)
    .build();

// 将特征数据写入 Redis Hash
DataStream<Tuple2<String, String>> featureStream = features
    .map(feature -> new Tuple2<>(
        "user_feature:" + feature.getUserId(),
        feature.toJsonString()
    ));

featureStream.addSink(new RedisSink<>(
    jedisConfig,
    new RedisCommandDescription(RedisCommand.HSET, "user_features")
));

// 设置 Sink 并行度，避免写入热点
featureStream.setParallelism(4);

3.6 三要素协同：构建完整作业

将 Source、Transformation、Sink 三要素串联起来，就构成了一个完整的 Flink 实时特征计算作业。以下是一个完整的作业骨架：

public class FeatureJob {
    public static void main(String[] args) throws Exception {
        // 1. 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        env.enableCheckpointing(5000);

        // 2. Source：从 Kafka 读取用户行为
        DataStream<String> rawStream = env
            .addSource(createKafkaSource())
            .name("kafka-source");

        // 3. Transformation：特征计算逻辑
        DataStream<UserFeature> featureStream = rawStream
            .map(new ParseFunction())
            .filter(new ValidFilter())
            .keyBy(UserClick::getUserId)
            .window(SlidingEventTimeWindows.of(Time.minutes(10), Time.minutes(1)))
            .aggregate(new FeatureAggregate())
            .name("feature-transform");

        // 4. Sink：输出特征到 Redis
        featureStream
            .map(f -> new Tuple2<>(f.getUserId(), f.toJson()))
            .addSink(new RedisSink<>(createRedisConfig(), new RedisCommandDescription(RedisCommand.SET)))
            .name("redis-sink");

        // 5. 触发执行
        env.execute("Real-time Feature Job");
    }
}

3.7 本章小结

本章系统讲解了 Flink DataStream API 的编程模型，重点掌握以下内容：

• StreamExecutionEnvironment：作业的入口和配置中心，负责并行度、Checkpoint、时间语义等核心设置。
• Source 三要素：Kafka Source 是实时特征计算的首选，务必配置 Watermark 策略以支持 EventTime 语义。
• Transformation 算子链：map/filter 做数据清洗，keyBy/window 做分组聚合，RichFunction 管理状态。
• Sink 输出：Redis 适合在线特征，Kafka 适合特征分发，选择时需权衡延迟、吞吐和一致性。
• 三要素协同：通过 env.execute() 触发作业，每个算子通过 .name() 标记便于运维。