Flink在EC2实例运行时无法读取S3文件

看起来你遇到的核心问题是Flink程序虽调用了env.execute()却仍返回提交异常，本质是无法正确读取S3上的文件。结合你的配置和代码，我整理了几个排查和解决方向：

1. 先确认S3路径和文件可访问性 #

• 检查S3路径写法：s3://CopyOfData/Days/作为目录路径，部分Flink版本需要明确指定通配符才能读取目录下所有文件，比如改成s3://CopyOfData/Days/*.csv（如果你的数据文件都是csv后缀）。
• 注意S3存储桶名称必须是小写（DNS兼容要求），如果你的桶名CopyOfData包含大写，会导致路径解析失败，建议改成小写桶名或者使用兼容的路径写法。
• 在EC2实例上直接用AWS CLI测试访问：

  # 列出目录下的文件
  aws s3 ls s3://CopyOfData/Days/
  # 下载一个文件到本地
  aws s3 cp s3://CopyOfData/Days/sample-file.csv ./
  

  如果这两步失败，说明IAM角色权限或路径有问题，先解决这个基础问题。

2. 验证IAM角色权限是否足够 #

EC2的IAM角色需要具备S3的列表桶和读取对象权限，确保你的角色策略包含以下内容：

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": "s3:ListBucket",
            "Resource": "arn:aws:s3:::CopyOfData"
        },
        {
            "Effect": "Allow",
            "Action": "s3:GetObject",
            "Resource": "arn:aws:s3:::CopyOfData/Days/*"
        }
    ]
}

如果之前的AWS CLI测试失败，大概率是权限不足，需要更新IAM角色策略。

3. 确保Flink加载了正确的Hadoop依赖和配置 #

虽然你设置了HADOOP_CONF_DIR，但Flink访问S3还需要对应的Hadoop AWS依赖：

• 下载与你的Hadoop版本兼容的hadoop-aws和aws-java-sdk-bundle（或拆分的SDK组件）JAR包，放到Flink安装目录的lib文件夹下。比如Hadoop 3.x对应hadoop-aws-3.x.x.jar，版本不匹配会导致文件系统初始化失败。
• 在flink-conf.yaml中添加S3文件系统的配置：

  fs.s3a.impl: org.apache.hadoop.fs.s3a.S3AFileSystem
  # 利用EC2 IAM角色获取凭证，无需手动配置access key
  fs.s3a.aws.credentials.provider: com.amazonaws.auth.InstanceProfileCredentialsProvider
  

• 确认HADOOP_CONF_DIR指向的目录中存在core-site.xml，并且文件中包含S3相关的配置（如果没有，上述flink-conf的配置也能生效）。

4. 查看Flink日志定位具体错误 #

客户端的错误提示比较笼统，建议查看Flink的详细日志：

• JobManager日志：log/flink-*-jobmanager-*.log，里面会记录作业提交时的初始化错误，比如S3连接失败、权限被拒等。
• TaskManager日志：log/flink-*-taskmanager-*.log，如果作业提交成功但TaskManager无法读取文件，这里会有具体的IOException信息。
  搜索日志中的关键词如S3、AccessDenied、FileNotFoundException，能快速定位问题根源。

最后再检查代码细节 #

确认你的readCsvFile调用是否符合预期：

• 如果目录下的文件格式不一致，可能需要添加更多配置（比如忽略首行、指定字段数量等）。
• 确保env是正确初始化的ExecutionEnvironment（批处理环境），你的代码用DataSet是没问题的，无需切换流处理环境。

内容的提问来源于stack exchange，提问作者Vijaya Seetharaman