4. 环境搭建与依赖：Flink集群部署与数据源集成

本章聚焦于构建PSI漂移告警系统所需的底层运行环境与依赖配置。我们将依次完成Flink集群的三种部署模式选型、Maven项目依赖的精确配置，以及Kafka/Pulsar两种主流消息队列的数据源集成。正确的环境搭建是后续实时计算任务稳定运行的基石。

4.1 Flink集群部署模式对比与选型

Flink支持多种部署模式，不同模式在资源管理、运维成本和弹性扩展上各有优劣。对于PSI漂移告警这类需要长期稳定运行的实时作业，选择合适的部署模式至关重要。

4.2 Flink on YARN 快速部署

以下以YARN模式为例，展示从环境准备到作业提交的完整流程。YARN模式能够充分利用集群资源，实现作业的自动容错与资源动态分配。

1. 前置条件： 已部署Hadoop集群（HDFS + YARN），版本建议为3.x。
2. 下载Flink： 从官网下载 flink-1.17.2-bin-scala_2.12.tgz 并解压。
3. 配置环境变量： 设置 HADOOP_CLASSPATH，确保Flink能加载Hadoop配置。
4. 启动YARN Session： 使用 ./bin/yarn-session.sh -d -jm 1024m -tm 2048m -s 2 启动一个常驻Session。
5. 提交作业： 通过 ./bin/flink run -m yarn-cluster -c com.example.PsiJob ./flink-psi-job.jar 提交PSI计算任务。

4.3 Maven依赖配置

一个典型的Flink PSI项目需要引入核心Flink依赖、连接器依赖以及PSI算法库。以下为 pom.xml 中的关键依赖配置片段。

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

<dependencies>
    <!-- Flink核心依赖 -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>${flink.version}</version>
        <scope>provided</scope>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java</artifactId>
        <version>${flink.version}</version>
        <scope>provided</scope>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-clients</artifactId>
        <version>${flink.version}</version>
        <scope>provided</scope>
    </dependency>

    <!-- Kafka连接器 -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-kafka</artifactId>
        <version>${flink.version}</version>
    </dependency>

    <!-- Pulsar连接器 -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-pulsar</artifactId>
        <version>4.0.0-1.17</version>
    </dependency>

    <!-- PSI算法库（示例） -->
    <dependency>
        <groupId>com.google.guava</groupId>
        <artifactId>guava</artifactId>
        <version>31.1-jre</version>
    </dependency>
</dependencies>

4.4 Kafka/Pulsar数据源集成

PSI漂移告警系统需要实时消费特征分布数据。Kafka和Pulsar是两种最常用的消息队列方案，Flink提供了原生连接器支持。

4.4.1 Kafka数据源集成

使用Flink Kafka Consumer消费数据，需要指定Topic、反序列化器和消费起始位置。

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

DataStream<String> stream = env.addSource(
    new FlinkKafkaConsumer<>(
        "feature-distribution-topic",
        new SimpleStringSchema(),
        kafkaProps
    )
);

4.4.2 Pulsar数据源集成

Pulsar连接器支持exactly-once语义和更灵活的订阅模式。以下为Pulsar Source的配置示例。

PulsarSource<String> pulsarSource = PulsarSource.builder()
    .setServiceUrl("pulsar://localhost:6650")
    .setAdminUrl("http://localhost:8080")
    .setStartCursor(StartCursor.earliest())
    .setTopics("persistent://public/default/feature-psi")
    .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(
        new SimpleStringSchema()))
    .setSubscriptionName("psi-subscription")
    .build();

DataStream<String> stream = env.fromSource(
    pulsarSource,
    WatermarkStrategy.noWatermarks(),
    "Pulsar Source"
);

4.5 数据源集成架构图

下图展示了PSI漂移告警系统中，Flink作业从Kafka/Pulsar消费数据，经过窗口聚合与PSI计算，最终输出告警信息的完整数据流架构。

4.6 环境验证与常见问题

完成部署与集成后，建议通过以下步骤验证环境是否就绪：

• 验证Flink集群： 访问Flink Web UI（默认端口8081），确认TaskManager和Slot状态正常。
• 验证Kafka连通性： 使用 kafka-console-producer 发送测试消息，在Flink作业中打印输出。
• 验证Pulsar连通性： 使用 pulsar-client 工具发送消息，确认Flink作业能正常消费。
• 检查依赖冲突： 使用 mvn dependency:tree 检查是否有重复或版本冲突的依赖。

至此，Flink集群环境、Maven依赖以及Kafka/Pulsar数据源集成已全部就绪。下一阶段将进入PSI算法的核心实现与窗口计算逻辑设计。