📚 课程目录第3章 DataStream API基础

第3章 DataStream API基础：执行环境、数据源、转换、数据汇与程序骨架

好，咱们进入实战前的第一道硬菜——DataStream API。说白了，这就是Flink流处理的“手脚架”。你写任何流作业，都绕不开这几个核心部件：执行环境、Source、Transformation、Sink。我刚开始学Flink时，总觉得这些概念太抽象，直到自己动手搭了一个实时同步任务，才真正理解它们是怎么串起来的。

3.1 执行环境（Execution Environment）

执行环境，就是你的Flink程序跑在哪个“舞台”上。它负责管理资源、调度任务、分配内存。我个人习惯，写任何Flink作业的第一步，就是先拿到这个环境对象。

Flink提供了三种执行环境：

类型	获取方式	适用场景
StreamExecutionEnvironment	`StreamExecutionEnvironment.getExecutionEnvironment()`	本地IDE调试或生产集群提交，自动识别
LocalStreamEnvironment	`StreamExecutionEnvironment.createLocalEnvironment()`	本地单机测试，方便断点调试
RemoteStreamEnvironment	`StreamExecutionEnvironment.createRemoteEnvironment(host, port, jarFiles)`	远程提交作业到Flink集群

我的小技巧：开发阶段我通常用getExecutionEnvironment()，它会根据你运行的环境自动切换。在IDE里跑就是本地模式，打成jar包提交到集群就是分布式模式。省心。

拿到环境后，你还可以设置一些全局参数，比如并行度、时间特性、检查点配置等。举个例子：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(4);                    // 设置全局并行度
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);  // 使用事件时间
env.enableCheckpointing(5000);            // 每5秒做一次检查点

嗯，这里要注意：并行度不是越大越好。我曾经在一个小集群上设了32并行度，结果资源争抢严重，反而比8并行度还慢。要根据你的数据量和集群规模来调。

3.2 数据源（Source）

数据源，就是数据从哪里来。Flink支持的数据源类型非常丰富，我把它分成三类：

基于文件：readTextFile()、readFile()，适合批处理或离线同步
基于Socket：socketTextStream()，测试时常用
基于消息队列：Kafka、RabbitMQ、Pulsar等，生产环境的主力

咱们做数据同步，最常用的就是Kafka Source。我给大家看一个典型的Kafka消费代码：

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

DataStream<String> stream = env
    .addSource(new FlinkKafkaConsumer<>(
        "sync-topic",
        new SimpleStringSchema(),
        props
    ));

避坑指南：我曾经犯过一个低级错误——Kafka消费端的group.id没设置，导致每次重启作业都从最新偏移量开始消费，丢了一堆数据。记住：生产环境一定要配group.id，并且要考虑消费起始位置。

除了内置Source，你还可以自定义Source。实现SourceFunction接口，或者继承RichSourceFunction。比如模拟数据生成器：

DataStream<String> mockStream = env.addSource(new RichSourceFunction<String>() {
    @Override
    public void run(SourceContext<String> ctx) throws Exception {
        while (isRunning) {
            ctx.collect("{\"id\":1, \"name\":\"test\"}");
            Thread.sleep(1000);
        }
    }
    @Override
    public void cancel() { isRunning = false; }
});

3.3 数据转换（Transformation）

数据转换，是Flink最核心的部分。说白了，就是把输入的数据流变成你想要的样子。常用的转换算子有这些：

算子	作用	我常用的场景
`map`	一对一转换，输入一条输出一条	字段映射、格式转换
`flatMap`	一对多转换，一条输入可输出0到多条	拆分JSON数组、日志解析
`filter`	过滤，满足条件才保留	过滤无效数据、脏数据清洗
`keyBy`	按key分组，后续可做聚合	按表名、数据库名分组同步
`reduce`	分组后的增量聚合	合并增量变更记录
`process`	最灵活的算子，可访问状态和定时器	复杂业务逻辑、延迟数据处理

举个例子，假设我们从Kafka拿到的是JSON字符串，需要解析成Java对象，然后过滤掉无效数据：

DataStream<UserEvent> parsedStream = kafkaStream
    .map(json -> JsonUtil.parse(json, UserEvent.class))
    .filter(event -> event != null && event.getUserId() > 0);

你想想看，如果不用filter，脏数据流到下游Sink，写入目标库时就会报错。我在做MySQL同步时，就遇到过因为空指针异常导致整个作业挂掉的情况。所以，数据清洗一定要在Transformation阶段做干净。

3.4 数据汇（Sink）

数据汇，就是处理完的数据往哪写。Flink内置了很多Sink，比如：

文件Sink：writeAsText()、StreamingFileSink
Socket Sink：writeToSocket()
JDBC Sink：JdbcSink，适合写入关系型数据库
Kafka Sink：FlinkKafkaProducer，适合写入消息队列

咱们做数据同步，最常用的是JDBC Sink和Kafka Sink。比如写入MySQL：

stream.addSink(JdbcSink.sink(
    "INSERT INTO user_events (user_id, event_type, event_time) VALUES (?, ?, ?)",
    (statement, event) -> {
        statement.setInt(1, event.getUserId());
        statement.setString(2, event.getEventType());
        statement.setTimestamp(3, event.getEventTime());
    },
    JdbcExecutionOptions.builder()
        .withBatchSize(1000)
        .withBatchIntervalMs(200)
        .build(),
    new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
        .withUrl("jdbc:mysql://localhost:3306/sync_db")
        .withDriverName("com.mysql.cj.jdbc.Driver")
        .withUsername("root")
        .withPassword("password")
        .build()
));

重点：JDBC Sink一定要设置批量写入参数。我见过有人一条一条插入，结果吞吐量只有几百条/秒。设置batchSize=1000后，轻松跑到几万条/秒。但也要注意，batchSize太大可能导致内存溢出，要根据你的数据大小来调。

3.5 Flink程序骨架

好了，前面四个部件都讲完了。把它们拼起来，就是一个完整的Flink程序。我总结了一个“四步法”骨架：

获取执行环境：StreamExecutionEnvironment env = ...
添加数据源：DataStream<T> source = env.addSource(...)
数据转换：DataStream<R> result = source.map(...).filter(...).keyBy(...)
输出数据汇：result.addSink(...)

最后别忘了调用env.execute("job name")来启动作业。我见过新手写了半天代码，结果忘了execute，程序跑完啥也没发生。

下面是一个完整的示例，把前面讲的内容串起来：

public class DataSyncJob {
    public static void main(String[] args) throws Exception {
        // 1. 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        env.enableCheckpointing(10000);

        // 2. 添加Kafka数据源
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "localhost:9092");
        props.setProperty("group.id", "sync-group");
        DataStream<String> source = env.addSource(
            new FlinkKafkaConsumer<>("input-topic", new SimpleStringSchema(), props)
        );

        // 3. 数据转换：解析JSON，过滤无效数据
        DataStream<SyncRecord> transformed = source
            .map(json -> JsonUtil.parse(json, SyncRecord.class))
            .filter(record -> record != null && record.isValid());

        // 4. 输出到MySQL
        transformed.addSink(createJdbcSink());

        // 5. 启动作业
        env.execute("Flink Data Sync Job");
    }
}

这个骨架，你以后写任何Flink流处理作业都可以套用。说白了，就是“取数据→处理数据→写数据”这三个步骤，Flink帮你把分布式、容错、并行这些复杂的事情都封装好了。

这张图就是Flink程序的“骨架”。你以后写任何作业，都可以照着这个结构来搭。记住：环境是舞台，Source是入口，Transformation是核心逻辑，Sink是出口。把这四个部件搞明白，DataStream API你就掌握了八成。