Flink无法从通过PipelineOptions添加的JAR中加载类。

可以使用 Flink 提供的 ClassPathConfigurer 类解决该问题。代码示例如下：

1. 在运行 Flink 程序时添加参数 -Dflink.class.path.dir=<JAR 文件夹路径>，指定 JAR 文件夹的路径。

2. 在 Flink 程序中使用 ClassPathConfigurer 类，将指定的 JAR 文件夹路径添加到 Flink 的 Classpath 中。

例如，以下代码将指定的 JAR 文件夹路径添加到 Flink 的 Classpath 中：

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.util.ClassPathConfigurer;

public class MySink extends RichSinkFunction<String> {

    private static final long serialVersionUID = 1L;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 获取参数 -Dflink.class.path.dir 的值
        String classPathDir = System.getProperty("flink.class.path.dir");
        // 将指定的 JAR 文件夹路径添加到 Flink 的 Classpath 中
        ClassPathConfigurer configurer = new ClassPathConfigurer(classPathDir);
        configurer.init(configuration);
    }

    @Override
    public void close() throws Exception {
    }
}