PyFlink读取S3桶数据失败求助（附代码及报错信息）

问题概述 #

尝试通过PyFlink创建数据流读取S3桶内文件时，触发UnsupportedFileSystemSchemeException，提示无法找到s3协议的文件系统实现。

环境信息 #

• PyFlink版本：apache-flink==1.19.0
• 使用的S3插件：flink-s3-fs-hadoop-1.19.0.jar

报错信息 #

Caused by: org.apache.flink.core.fs.UnsupportedFileSystemSchemeException: Could not find a file system implementation for scheme 's3'. The scheme is directly supported by Flink through the following plugin(s): flink-s3-fs-hadoop, flink-s3-fs-presto. Please ensure that each plugin resides within its own subfolder within the plugins directory. See https://nightlies.apache.org/flink/flink-docs-stable/docs/deployment/filesystems/plugins/ for more information. If you want to use a Hadoop file system for that scheme, please add the scheme to the configuration fs.allowed-fallback-filesystems. For a full list of supported file systems, please see https://nightlies.apache.org/flink/flink-docs-stable/ops/filesystems/.

解决步骤 #

1. 修正插件加载方式 #

Flink的文件系统插件必须放在单独的子目录下，而非通过代码中的pipeline.jars参数加载：

• 找到你的Flink安装目录，在plugins文件夹下创建s3-fs-hadoop子目录
• 将flink-s3-fs-hadoop-1.19.0.jar放入该子目录中

2. 调整代码配置 #

移除代码中通过pipeline.jars加载插件的行，修正后的代码如下：

import os
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.common import Configuration

def main():
    config = Configuration()
    aws_access_key_id = 'aws_access_key_id'
    aws_secret_access_key = 'aws_secret_access_key'

    # 移除原有的pipeline.jars配置
    config.set_string("python.executable", "python3")
    config.set_string("fs.s3.access-key", aws_access_key_id)
    config.set_string("fs.s3.secret-key", aws_secret_access_key)
    config.set_boolean("fs.s3.path-style-access", True)
    config.set_string("fs.allowed-fallback-filesystems", "s3")
    
    env = StreamExecutionEnvironment.get_execution_environment(config)
    env.set_parallelism(1)

    data_stream = env.read_text_file("s3://historic-data/2023/01/01/")
    data_stream.print()
    env.execute("Read S3 Data Job")

if __name__ == '__main__':
    main()

注意：AWS S3无需手动指定fs.s3.endpoint，Flink会自动适配；若使用S3兼容存储（如MinIO），才需要配置对应endpoint。

3. 验证插件加载 #

确保Flink启动时能扫描到s3-fs-hadoop插件目录。本地运行PyFlink时，Flink会自动读取安装目录下的plugins文件夹；集群部署时，需保证所有节点的plugins目录结构一致。

4. 权限校验 #

确认你的AWS密钥拥有目标S3桶的GetObject、ListBucket等权限，避免因权限不足导致隐性错误。

内容的提问来源于stack exchange，提问作者Sumit Kashyap