2. Flink 环境搭建与数据源接入

说实话，很多初学者一上来就盯着 Flink 的复杂 API 看，结果环境都没跑通，心态先崩了。我个人习惯是先把开发环境收拾利索，再谈什么脏数据清洗。这一章，咱们就把 Flink 的开发环境搭起来，然后把最常见的三种数据源——Kafka、文件、Socket——都接进来。你跟着我走一遍，后面写代码心里就有底了。

2.1 开发环境配置：从零到能跑

我见过不少同事在环境配置上浪费一整天。其实没那么复杂，核心就三步：装 JDK、配 Maven、写个 WordCount 验证。嗯，这里要注意版本匹配。

2.1.1 基础依赖与版本选择

Flink 对 Java 版本有要求。我个人建议用 JDK 8 或 JDK 11，别追新。为什么？因为 Flink 1.13 到 1.16 对 JDK 8 支持最好，生产环境也最稳。我曾经在项目里试过 JDK 17，结果遇到一些序列化兼容问题，折腾了半天才回退。

组件	推荐版本	说明
JDK	8 / 11	建议用 JDK 8，兼容性最好
Maven	3.6+	用于项目构建和依赖管理
Flink	1.14 / 1.15	稳定版，社区活跃
IDE	IntelliJ IDEA	社区版即可，装好 Scala 插件

2.1.2 Maven 项目骨架与依赖

创建一个 Maven 项目，pom.xml 里加上 Flink 的核心依赖。我习惯把 Flink 版本号抽成属性，方便以后升级。

<properties>
    <flink.version>1.15.2</flink.version>
    <scala.binary.version>2.12</scala.binary.version>
</properties>

<dependencies>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-clients</artifactId>
        <version>${flink.version}</version>
    </dependency>
</dependencies>

2.1.3 验证环境：跑一个简单的 WordCount

环境配好之后，我建议你立刻跑一个 WordCount 验证。别嫌它简单，它能帮你确认依赖有没有冲突、网络通不通、本地环境能不能启动 JobManager。

public class WordCount {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

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

        DataStream<Tuple2<String, Integer>> counts = text
            .flatMap((String line, Collector<Tuple2<String, Integer>> out) -> {
                for (String word : line.split(" ")) {
                    out.collect(new Tuple2<>(word, 1));
                }
            })
            .returns(Types.TUPLE(Types.STRING, Types.INT))
            .keyBy(value -> value.f0)
            .sum(1);

        counts.print();

        env.execute("Socket WordCount");
    }
}

启动后，在终端执行 nc -lk 9999，输入几个单词，看控制台有没有输出。有输出，说明环境没问题。我曾经有一次怎么都跑不起来，最后发现是防火墙把 9999 端口封了……你想想看，这种低级错误排查起来最费时间。

2.2 读取 Kafka 数据源：生产环境标配

Kafka 是 Flink 最常用的数据源，没有之一。你在生产环境里，八成以上的 Flink 任务都是从 Kafka 读数据。脏数据也大多从这里混进来。

2.2.1 添加 Kafka Connector 依赖

Flink 官方提供了 Kafka connector，但要注意版本匹配。我建议用 flink-connector-kafka，别用老版的 flink-kafka-connector，后者已经 deprecated 了。

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka</artifactId>
    <version>${flink.version}</version>
</dependency>

2.2.2 创建 Kafka Source

核心代码就几行。你需要指定 topic、bootstrap servers、反序列化方式。我习惯用 SimpleStringSchema 先拿到原始字符串，后面再根据业务做解析。

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

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

2.2.3 反序列化与脏数据预处理

Kafka 里的数据不一定是干净的 JSON。我遇到过字段缺失、类型错误、甚至乱码的情况。所以建议在 Source 之后立刻加一个 map 或 flatMap 做初步清洗。

DataStream<MyEvent> cleanStream = kafkaStream
    .flatMap(new FlatMapFunction<String, MyEvent>() {
        @Override
        public void flatMap(String value, Collector<MyEvent> out) {
            try {
                MyEvent event = JsonUtil.parse(value, MyEvent.class);
                if (event != null && event.isValid()) {
                    out.collect(event);
                }
            } catch (Exception e) {
                // 脏数据直接丢弃，或者写入侧输出流
                // 这里先不展开，后面章节会详细讲
            }
        }
    });

2.3 读取文件数据源：离线与测试场景

文件数据源在两种场景下很常用：一是离线批处理，二是本地测试。你想想看，开发阶段总不能每次都连 Kafka 吧？用文件数据源调试方便多了。

2.3.1 读取本地文件

Flink 支持读取本地文件系统、HDFS、S3 等。最简单的用法是 readTextFile。

DataStream<String> fileStream = env.readTextFile("file:///path/to/input.txt");

如果你要读取整个目录，或者按行处理 CSV、JSON 文件，可以用 FileInputFormat 自定义。我建议用 readFile 方法，它支持监控文件变化，适合持续读取的场景。

DataStream<String> fileStream = env.readFile(
    new TextInputFormat(new Path("file:///path/to/dir")),
    "file:///path/to/dir",
    FileProcessingMode.PROCESS_CONTINUOUSLY,
    1000
);

2.3.2 处理文件中的脏数据

文件数据源里的脏数据通常表现为：空行、分隔符不一致、编码问题。我建议在读取后立即做一次过滤。

DataStream<String> filteredStream = fileStream
    .filter(line -> line != null && !line.trim().isEmpty())
    .map(line -> line.replaceAll("\\r|\\n", "")); // 去掉换行符干扰

2.4 读取 Socket 数据源：快速原型验证

Socket 数据源虽然不用于生产，但它是验证 Flink 逻辑最快的方式。你不需要 Kafka、不需要文件，一个 nc 命令就能搞定。

2.4.1 使用 socketTextStream

Flink 提供了 socketTextStream 方法，直接指定 host 和 port 就行。

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

就这么简单。启动后，你在终端输入什么，Flink 就收到什么。我经常用它来测试清洗逻辑——比如输入一些格式错误的数据，看清洗算子能不能正确处理。

2.5 本章知识体系总览

下面这张图把本章的核心内容串起来了。你可以看到，三种数据源最终都汇入 Flink 的 DataStream API，然后进入清洗流程。

2.6 小结与避坑清单

这一章的内容其实不难，但细节多。我把常见的坑列出来，你对照着检查，能省不少时间。

• 版本匹配： Flink、Kafka connector、JDK 版本必须兼容。我建议用 Flink 1.15 + JDK 8 + Kafka 2.8。
• 依赖冲突： 尤其是 Kafka client 版本，Flink 自带的可能和你项目里的冲突。用 mvn dependency:tree 检查。
• 网络连通性： Socket 和 Kafka 都需要网络可达。本地测试时注意防火墙和端口占用。
• 数据源选择： 生产用 Kafka，测试用文件或 Socket。别搞混了。