Flink 外部资源框架：作业原生申请 GPU/FPGA 资源

一个包含 GPU+FPGA 的示例：

external-resources: gpu;fpga
external-resource.gpu.driver-factory.class: org.apache.flink.externalresource.gpu.GPUDriverFactory
external-resource.gpu.amount: 2
external-resource.gpu.param.discovery-script.args: --enable-coordination-mode
external-resource.fpga.driver-factory.class: org.apache.flink.externalresource.fpga.FPGADriverFactory
external-resource.fpga.amount: 1
external-resource.fpga.yarn.config-key: yarn.io/fpga

3.3 在算子里使用 RuntimeContext 获取资源信息

算子侧用法非常直接：

public class ExternalResourceMapFunction extends RichMapFunction<String, String> {
    private static final String RESOURCE_NAME = "gpu";

    @Override
    public String map(String value) throws Exception {
        Set<ExternalResourceInfo> infos = getRuntimeContext().getExternalResourceInfos(RESOURCE_NAME);
        List<String> indices = new ArrayList<>();
        for (ExternalResourceInfo info : infos) {
            info.getProperty("index").ifPresent(indices::add); // GPU 插件常用属性 key：index
        }
        // 这里用 indices 做设备绑定，比如选择一张卡 set CUDA_VISIBLE_DEVICES 或初始化推理引擎
        return value;
    }
}

每个 ExternalResourceInfo 里有哪些 key，取决于插件实现。你可以用 ExternalResourceInfo#getKeys() 获取完整键集合。

4. Kubernetes / YARN / Standalone：三种环境的差异

4.1 Kubernetes

• K8s 原生通过 Device Plugin 机制提供 GPU/FPGA 等资源（Kubernetes v1.10+ 支持）
• Flink 会把你配置的 kubernetes.config-key 写入 TaskManager 主容器的：
  • resources.requests.<config-key>
  • resources.limits.<config-key>

GPU 的常见 key：

• NVIDIA：nvidia.com/gpu
• AMD：amd.com/gpu（但 Flink 默认 discovery 脚本是 NVIDIA 的，AMD 需要你自己写脚本）

4.2 YARN

• YARN 2.10+ / 3.1+ 开始支持 GPU/FPGA 资源
• Flink 通过 external-resource.<name>.yarn.config-key 把 amount 写进 container resource profile

GPU（YARN）常见 key：

• yarn.io/gpu（注意：YARN 当前通常仅支持 NVIDIA GPU 的调度）

4.3 Standalone

• Standalone 模式没有底层 RM 帮你'保证资源'，你需要管理员在节点上确保外部资源可用（驱动安装、权限、设备可见性等）
• 如果同一台机器上跑多个 TaskManager，GPU 可见性很容易冲突，需要配合 discovery script 的协调模式

5. GPU 插件：最常用也最值得踩坑的一块

Flink 目前官方提供的一方外部资源插件就是 GPU 插件。

5.1 必要配置（GPU）

external-resources: gpu
external-resource.gpu.driver-factory.class: org.apache.flink.externalresource.gpu.GPUDriverFactory
external-resource.gpu.amount: 2
# Kubernetes
external-resource.gpu.kubernetes.config-key: nvidia.com/gpu
# YARN
external-resource.gpu.yarn.config-key: yarn.io/gpu

5.2 discovery script（GPU 发现脚本）

GPUDriver 会调用 discovery script 来发现可用 GPU，并生成 ExternalResourceInfo，其中关键属性是：

• key = index（GPU 设备 index）

默认脚本路径（NVIDIA）：

external-resource.gpu.param.discovery-script.path: plugins/external-resource-gpu/nvidia-gpu-discovery.sh

自定义脚本路径（例如 AMD）也可以配置同一个 key。

脚本参数：

external-resource.gpu.param.discovery-script.args: --enable-coordination-mode

5.3 脚本契约（你写自定义脚本时必须遵守）

• Flink 先把 amount 作为第一个参数传给脚本
• 你配置的 discovery-script.args 会拼在后面
• 脚本输出：用逗号分隔的 GPU index 列表，例如 0,1
• 输出里纯空白 index 会被忽略
• 如果发现失败或数量不足：脚本返回非 0 exit code，Flink 将不会向算子提供 gpu 信息（RuntimeContext 拿不到）

5.4 协调模式：解决'同机多 TM 抢同一 GPU'

Standalone 下经常同机部署多个 TaskManager，此时所有 TM 默认都能看到同一批 GPU（nvidia-smi 可见），很容易'多进程抢同一张卡'。

默认脚本提供 coordination mode：

• --enable-coordination-mode：启用协调
• --coordination-file <path>：协调文件路径（默认 /var/tmp/flink-gpu-coordination）

它能保证：同一个 Flink 集群内，同机多个 TaskManager 不会分到同一张 GPU。

但要注意两点：

• 只在'同一协调文件范围内'有效：另一个 Flink 集群如果用不同 coordination file，仍可能抢同一 GPU
• 非 Flink 应用也可能用同一 GPU，这个模式无法防住

6. 自定义资源插件：你要支持 FPGA/自研加速卡怎么做

你需要实现三件套：

1）ExternalResourceDriver

• retrieveResourceInfo(long amount)：返回 ExternalResourceInfo 集合（你定义的资源维度）

2）ExternalResourceDriverFactory

• createExternalResourceDriver(Configuration config)：从配置创建 driver

3）SPI 服务声明（非常关键）

• 在 jar 内创建文件：
  • META-INF/services/org.apache.flink.api.common.externalresource.ExternalResourceDriverFactory
• 文件内容写你的 factory 全类名，例如：
  • your.domain.FPGADriverFactory

示例骨架：

public class FPGAInfo implements ExternalResourceInfo {
    @Override
    public Optional<String> getProperty(String key) {
        // 根据 key 返回属性，比如 "device", "pci", "address" 等
        return Optional.empty();
    }

    @Override
    public Collection<String> getKeys() {
        return List.of("device", "pci", "address");
    }
}

public class FPGADriver implements ExternalResourceDriver {
    @Override
    public Set<ExternalResourceInfo> retrieveResourceInfo(long amount) {
        // 发现并返回 FPGA 信息集合
        return Set.of(/* ... */);
    }
}

public class FPGADriverFactory implements ExternalResourceDriverFactory {
    @Override
    public ExternalResourceDriver createExternalResourceDriver(Configuration config) {
        return new FPGADriver();
    }
}

打包成 jar 丢到 plugins/fpga/，然后在 flink-conf.yaml 里按 <resource_name> 配置启用即可。

7. 排障清单：最常见的 6 个'为什么拿不到 GPU'

1）external-resources 没写 gpu（或资源名拼错） 2）插件 jar 没放到 plugins/ 下正确目录（或目录里缺依赖） 3）没配置 driver-factory.class，导致算子侧拿不到 ExternalResourceInfo 4）K8s 没装 NVIDIA device plugin（Pod 根本拿不到 GPU） 5）discovery script 不可执行 / 路径不对 / 返回非 0 6）Standalone 同机多 TM 没开协调模式，导致资源冲突看似'有卡但不可用'

8. 最佳实践建议

• 先明确'资源申请'和'资源绑定'是两步：申请解决'容器是否带卡'，绑定解决'算子用哪张卡'
• GPU 推理算子里要做设备亲和：基于 index 决定 CUDA_VISIBLE_DEVICES 或引擎初始化参数
• Standalone 同机多 TM 建议默认开 coordination mode
• 生产上尽量用 Kubernetes/YARN 去做资源保证，Standalone 只适合可控环境
• 由于没有 operator 级隔离，最好避免在同一 TM 内多个算子'各自随便挑卡'，要统一策略（例如只由一个算子管理 GPU worker）