【发布时间】:2019-06-21 22:08:56
【问题描述】:
我正在开发一个将一些文件上传到 s3 存储桶的应用程序,稍后,它会从 s3 存储桶中读取文件并将其推送到我的数据库 .
我正在使用 Flink 1.4.2 和 fs.s3a API 从 s3 存储桶读取和写入文件。
将文件上传到 s3 存储桶可以正常工作,没有任何问题,但是当我的应用程序从 s3 读取这些上传文件的第二阶段启动时,我的应用程序抛出以下错误:
Caused by: java.io.InterruptedIOException: Reopen at position 0 on s3a://myfilepath/a/b/d/4: org.apache.flink.fs.s3hadoop.shaded.com.amazonaws.SdkClientException: Unable to execute HTTP request: Timeout waiting for connection from pool
at org.apache.flink.fs.s3hadoop.shaded.org.apache.hadoop.fs.s3a.S3AUtils.translateException(S3AUtils.java:125)
at org.apache.flink.fs.s3hadoop.shaded.org.apache.hadoop.fs.s3a.S3AInputStream.reopen(S3AInputStream.java:155)
at org.apache.flink.fs.s3hadoop.shaded.org.apache.hadoop.fs.s3a.S3AInputStream.lazySeek(S3AInputStream.java:281)
at org.apache.flink.fs.s3hadoop.shaded.org.apache.hadoop.fs.s3a.S3AInputStream.read(S3AInputStream.java:364)
at java.io.DataInputStream.read(DataInputStream.java:149)
at org.apache.flink.fs.s3hadoop.shaded.org.apache.flink.runtime.fs.hdfs.HadoopDataInputStream.read(HadoopDataInputStream.java:94)
at org.apache.flink.api.common.io.DelimitedInputFormat.fillBuffer(DelimitedInputFormat.java:702)
at org.apache.flink.api.common.io.DelimitedInputFormat.open(DelimitedInputFormat.java:490)
at org.apache.flink.api.common.io.GenericCsvInputFormat.open(GenericCsvInputFormat.java:301)
at org.apache.flink.api.java.io.CsvInputFormat.open(CsvInputFormat.java:53)
at org.apache.flink.api.java.io.PojoCsvInputFormat.open(PojoCsvInputFormat.java:160)
at org.apache.flink.api.java.io.PojoCsvInputFormat.open(PojoCsvInputFormat.java:37)
at org.apache.flink.runtime.operators.DataSourceTask.invoke(DataSourceTask.java:145)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:718)
at java.lang.Thread.run(Thread.java:748)
我能够通过增加 s3a API 的最大连接数参数来控制此错误。
截至目前,我在 s3 存储桶中有大约 1000 个文件,这些文件由我的应用在 s3 存储桶中推送和拉取,我的最大连接数是3000。我正在使用 Flink 的并行性从 s3 存储桶上传/下载这些文件。我的任务管理器计数为 14。 这是间歇性失败,我也有这种情况的成功案例。
我的查询是,
- 为什么会出现间歇性故障?如果我设置的最大连接数很低,那么我的应用程序应该在每次运行时都会抛出此错误。
- 有没有什么方法可以计算出我的应用程序运行所需的最佳最大连接数,而不会遇到连接池超时错误?或者这个错误是否与我不知道的其他事情有关?
谢谢 提前
【问题讨论】:
标签: amazon-s3 apache-flink flink-streaming