4. Flink DataStream API 基础：Source、Transformation、Sink

本章是 Flink 实时额度计算实践的数据接入与处理基础。你将掌握 DataStream API 三大核心组件：Source（数据源）、Transformation（转换算子）和 Sink（数据输出）。在额度计算场景中，Source 负责从 Kafka 或 Socket 读取交易流水，Transformation 执行过滤、映射与聚合，Sink 将结果写回 Kafka 或打印到控制台。

4.1 Source：数据从哪里来？

Source 是 Flink 作业的数据入口。额度计算通常需要实时接入交易事件，以下三种 Source 覆盖了开发调试与生产部署场景。

4.1.1 集合 Source

最轻量的数据源，适用于本地验证算子逻辑。额度计算中常用 fromCollection 模拟交易事件流。

// 模拟三条交易记录
DataStream<String> sourceStream = env.fromCollection(
    Arrays.asList(
        "user_001,100.50,PAY",
        "user_002,200.00,RECHARGE",
        "user_001,50.25,PAY"
    )
);

4.1.2 Socket Source

通过 socketTextStream 监听指定主机和端口，适合在开发环境模拟持续流入的数据。

DataStream<String> socketStream = env.socketTextStream("localhost", 9999);

4.1.3 Kafka Source（生产首选）

Flink 与 Kafka 集成是实时计算的标准架构。额度计算场景下，Kafka 主题通常存储交易事件、账户变更或风控指令。

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

DataStream<String> kafkaStream = env.addSource(
    new FlinkKafkaConsumer<>(
        "trade-events",          // 主题名
        new SimpleStringSchema(), // 反序列化器
        kafkaProps
    )
);

4.2 Transformation：核心算子

Transformation 是 DataStream API 的“加工车间”。额度计算中，map 用于格式转换，flatMap 用于拆分与展开，filter 用于剔除无效或超限交易。

4.2.1 map：一对一转换

将输入元素转换为另一个元素，常用于解析字符串、提取字段或格式化输出。额度计算中，将 CSV 交易记录转为 Java 对象。

// 将 "user_001,100.50,PAY" 解析为 TradeEvent 对象
DataStream<TradeEvent> tradeStream = sourceStream.map(new MapFunction<String, TradeEvent>() {
    @Override
    public TradeEvent map(String value) throws Exception {
        String[] fields = value.split(",");
        return new TradeEvent(
            fields[0],                    // userId
            Double.parseDouble(fields[1]), // amount
            fields[2]                     // type
        );
    }
});

4.2.2 flatMap：一对多转换

一个输入元素可以产生零个、一个或多个输出元素。在额度计算中，用于将一笔复合交易拆分为多条子交易。

// 将一笔包含多个商品的订单拆分为多条子交易
DataStream<SubTrade> subTradeStream = tradeStream.flatMap(new FlatMapFunction<TradeEvent, SubTrade>() {
    @Override
    public void flatMap(TradeEvent event, Collector<SubTrade> out) throws Exception {
        for (String item : event.getItems()) {
            out.collect(new SubTrade(event.getUserId(), item, event.getAmount() / event.getItems().size()));
        }
    }
});

4.2.3 filter：过滤保留

根据条件保留或丢弃元素。额度计算中，过滤掉金额为负或超过单笔限额的交易。

// 只保留 PAY 类型且金额在 0~10000 之间的交易
DataStream<TradeEvent> validStream = tradeStream.filter(new FilterFunction<TradeEvent>() {
    @Override
    public boolean filter(TradeEvent event) throws Exception {
        return "PAY".equals(event.getType())
            && event.getAmount() > 0
            && event.getAmount() <= 10000;
    }
});

4.3 Sink：结果写到哪里？

Sink 定义了处理结果的输出目的地。额度计算中，print 用于调试，Kafka Sink 用于将计算结果写回消息队列供下游消费。

4.3.1 print Sink

最简单的 Sink，将每个元素打印到 TaskManager 的标准输出。每个并行实例会添加前缀（如 1>）。

validStream.print("额度计算结果");

4.3.2 Kafka Sink（生产标准）

将处理后的额度变更事件写回 Kafka，供风控系统、告警服务或实时大屏消费。

Properties producerProps = new Properties();
producerProps.setProperty("bootstrap.servers", "localhost:9092");
producerProps.setProperty("retries", "3");

validStream.addSink(new FlinkKafkaProducer<>(
    "quota-result",              // 目标 Topic
    new SimpleStringSchema(),    // 序列化器
    producerProps
));

4.4 完整额度计算管道示例

将本章知识点串联：从 Kafka 读取交易事件 → 解析为对象 → 过滤无效交易 → 写回 Kafka。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(5000); // 5秒一次 Checkpoint

// 1. Source：从 Kafka 读取
Properties consumerProps = new Properties();
consumerProps.setProperty("bootstrap.servers", "localhost:9092");
consumerProps.setProperty("group.id", "quota-group");

DataStream<String> rawStream = env.addSource(
    new FlinkKafkaConsumer<>("trade-events", new SimpleStringSchema(), consumerProps)
);

// 2. Transformation：解析 + 过滤
DataStream<TradeEvent> validStream = rawStream
    .map(new MapFunction<String, TradeEvent>() {
        @Override
        public TradeEvent map(String value) throws Exception {
            String[] fields = value.split(",");
            return new TradeEvent(fields[0], Double.parseDouble(fields[1]), fields[2]);
        }
    })
    .filter(new FilterFunction<TradeEvent>() {
        @Override
        public boolean filter(TradeEvent event) throws Exception {
            return event.getAmount() > 0 && event.getAmount() <= 10000;
        }
    });

// 3. Sink：写回 Kafka
Properties producerProps = new Properties();
producerProps.setProperty("bootstrap.servers", "localhost:9092");
validStream.addSink(new FlinkKafkaProducer<>("quota-result", new SimpleStringSchema(), producerProps));

env.execute("Quota Calculation Pipeline");