3. Flink 环境搭建：集群安装、开发环境配置与作业提交

在交易风控场景中，Flink 集群的稳定性和开发环境的正确配置直接决定了样本切分任务的实时性与准确性。本章将围绕 Flink 集群的安装部署、开发环境搭建以及作业提交方式展开，确保你能够快速上手并投入实战。

3.1 Flink 集群安装

Flink 支持多种部署模式，包括 Standalone、YARN、Kubernetes 等。在风控样本切分场景中，Standalone 模式常用于开发测试，YARN 模式则更适合生产环境。以下以 Standalone 模式为例，展示集群安装的核心步骤。

3.1.1 环境准备

• JDK 1.8+：Flink 依赖 Java 运行环境，建议使用 JDK 8 或 11。
• SSH 免密登录：集群节点间需要无密码通信。
• 时钟同步：使用 NTP 确保所有节点时间一致，避免事件时间窗口错乱。

3.1.2 安装步骤

1. 下载 Flink 安装包（以 Flink 1.17 为例）并解压至 /opt/flink。
2. 配置 conf/flink-conf.yaml，设置 jobmanager.rpc.address 为 Master 节点 IP。
3. 配置 conf/workers 文件，添加所有 TaskManager 节点的主机名。
4. 将安装目录分发至所有 Worker 节点。
5. 在 Master 节点执行 bin/start-cluster.sh 启动集群。

3.1.3 集群架构示意

下图展示了 Standalone 模式下 Flink 集群的核心组件与通信关系：

3.2 Flink 开发环境配置

开发环境是编写和调试风控样本切分逻辑的基础。推荐使用 IntelliJ IDEA + Maven 作为主力开发工具，并配合 Flink 提供的 Maven 原型快速创建项目骨架。

3.2.1 项目依赖配置

在 pom.xml 中引入 Flink 核心依赖，注意版本与集群保持一致：

<properties>
    <flink.version>1.17.0</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>
        <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>
</dependencies>

3.2.2 本地运行模式

在 IDE 中直接运行 Flink 作业时，会自动启动一个 MiniCluster 用于调试。以下是一个简单的风控样本切分入口示例：

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

        DataStream<String> source = env.socketTextStream("localhost", 9999);
        source
            .map(line -> line.split(","))
            .filter(fields -> fields.length >= 5)
            .map(fields -> new Transaction(fields[0], Double.parseDouble(fields[4])))
            .keyBy(Transaction::getUserId)
            .window(TumblingEventTimeWindows.of(Time.minutes(1)))
            .process(new RiskWindowFunction())
            .print();

        env.execute("Risk Sample Split Job");
    }
}

3.3 Flink 作业提交方式

Flink 支持多种作业提交方式，适用于不同阶段和场景。下表对比了三种主流提交方式的核心差异：

3.3.1 命令行提交实战

以 Standalone 集群为例，提交风控样本切分作业的命令如下：

# 提交作业到远程集群
./bin/flink run \
  -m 192.168.1.100:8081 \
  -p 8 \
  -c com.risk.sample.RiskSampleJob \
  /opt/jars/risk-sample-1.0.jar \
  --input kafka://transactions \
  --output hdfs:///risk/samples

3.3.2 作业提交流程示意

下图展示了从客户端提交作业到 Flink 集群的完整流程：

3.4 环境验证与常见问题

完成搭建后，建议通过以下清单进行验证：

• Web UI 可达：浏览器访问 http://<master>:8081，确认集群状态为 Running。
• 作业提交测试：运行一个简单的 WordCount 示例，观察 Task 分配与数据流。
• 日志检查：查看 log/flink-*-jobmanager-*.log，确认无异常堆栈。