使用Flink 1.16.0 Table API读取S3 Ceph Parquet文件遇依赖错误求助

问题原因 #

java.lang.NoClassDefFoundError: org/apache/hadoop/conf/Configuration 是因为Flink的FileSystem连接器在处理Ceph这类S3兼容对象存储时，依赖Hadoop的核心配置类及S3适配实现，但现有依赖中缺少完整的Hadoop相关包。

解决方案 #

1. 补充正确的Hadoop依赖 #

在pom.xml中添加与Flink 1.16兼容的Hadoop核心和S3适配依赖（推荐Hadoop 3.2.x系列）：

<!-- 提供Configuration类的Hadoop核心依赖 -->
<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-common</artifactId>
    <version>3.2.4</version>
    <!-- 本地调试用compile；提交到已部署Hadoop的Flink集群时改为provided -->
    <scope>compile</scope>
</dependency>

<!-- 用于连接Ceph S3的Hadoop S3适配依赖 -->
<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-aws</artifactId>
    <version>3.2.4</version>
    <scope>compile</scope>
</dependency>

2. 配置Ceph S3连接参数 #

以Table API为例，在代码中配置Ceph的S3端点及认证信息：

TableEnvironment tableEnv = TableEnvironment.create(EnvironmentSettings.inStreamingMode());

// 配置Ceph S3连接参数
Configuration tableConf = tableEnv.getConfig().getConfiguration();
tableConf.setString("fs.s3a.endpoint", "http://your-ceph-s3-endpoint:port");
tableConf.setString("fs.s3a.access.key", "your-access-key");
tableConf.setString("fs.s3a.secret.key", "your-secret-key");
// Ceph需启用路径风格访问，禁用虚拟主机模式
tableConf.setString("fs.s3a.path.style.access", "true");
tableConf.setString("fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem");

// 定义读取Parquet文件的表
tableEnv.executeSql("CREATE TABLE ceph_parquet_table (" +
    "  id INT," +
    "  name STRING," +
    "  ts TIMESTAMP" +
    ") WITH (" +
    "  'connector' = 'filesystem'," +
    "  'path' = 's3a://your-bucket/path/to/parquet'," +
    "  'format' = 'parquet'" +
    ")");

// 执行查询并输出结果
tableEnv.executeSql("SELECT * FROM ceph_parquet_table").print();

3. 注意事项 #

• 若向Flink集群提交作业，确保集群flink/lib目录包含Hadoop相关包，或通过-yD参数指定依赖，避免版本冲突。
• 保持所有Hadoop依赖版本一致，不要混合不同版本的Hadoop包，否则可能引发其他类加载异常。

内容的提问来源于stack exchange，提问作者Lior Livshits