You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user-zh@flink.apache.org by 阿华田 <a1...@163.com> on 2020/06/01 06:48:32 UTC
FileInputFormat 使用问题
使用FileInputFormat 递归读取hdfs文件，并添加过滤器。程序执行没有报错但是很快就执行完成也没有读取到数据，本地测试可以过滤并读取到数据，yarn集群上执行出现上述情况。
代码：
//初始化任务参数
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
FileInputFormat fileInputFormat = new TextInputFormat(new Path("hdfs://arc/success_fid_flow"));
fileInputFormat.setNestedFileEnumeration(true);
//过滤掉条件为true
fileInputFormat.setFilesFilter(new RegexExcludePathAndTimeFilter("2020-05-24","2020-05-24"));
DataSet<String> source =env.createInput(fileInputFormat);
source.output(new HdfsTrainSinktest());
打印的日志:
2020-06-01 14:43:41,848 INFO org.apache.flink.configuration.GlobalConfiguration - Loading configuration property: jobmanager.rpc.address, localhost
2020-06-01 14:43:41,848 INFO org.apache.flink.configuration.GlobalConfiguration - Loading configuration property: jobmanager.rpc.port, 6123
2020-06-01 14:43:41,848 INFO org.apache.flink.configuration.GlobalConfiguration - Loading configuration property: jobmanager.heap.size, 1024m
2020-06-01 14:43:41,848 INFO org.apache.flink.configuration.GlobalConfiguration - Loading configuration property: taskmanager.heap.size, 1024m
2020-06-01 14:43:41,849 INFO org.apache.flink.configuration.GlobalConfiguration - Loading configuration property: taskmanager.numberOfTaskSlots, 1
2020-06-01 14:43:41,849 INFO org.apache.flink.configuration.GlobalConfiguration - Loading configuration property: parallelism.default, 1
2020-06-01 14:43:41,849 INFO org.apache.flink.configuration.GlobalConfiguration - Loading configuration property: jobmanager.execution.failover-strategy, region
2020-06-01 14:43:41,849 INFO org.apache.flink.configuration.GlobalConfiguration - Loading configuration property: historyserver.web.address, 0.0.0.0
2020-06-01 14:43:41,849 INFO org.apache.flink.configuration.GlobalConfiguration - Loading configuration property: historyserver.web.port, 8082
2020-06-01 14:43:41,890 INFO org.apache.flink.api.java.ExecutionEnvironment - The job has 0 registered types and 0 default Kryo serializers
2020-06-01 14:43:41,993 INFO org.apache.flink.optimizer.Optimizer - Compiler could not determine the size of input 'TextInputFormat ([hdfs://arc/success_fid_flow]) - UTF-8'. Using default estimates.
2020-06-01 14:43:42,022 INFO org.apache.flink.configuration.GlobalConfiguration - Loading configuration property: jobmanager.rpc.address, localhost
2020-06-01 14:43:42,022 INFO org.apache.flink.configuration.GlobalConfiguration - Loading configuration property: jobmanager.rpc.port, 6123
2020-06-01 14:43:42,022 INFO org.apache.flink.configuration.GlobalConfiguration - Loading configuration property: jobmanager.heap.size, 1024m
2020-06-01 14:43:42,022 INFO org.apache.flink.configuration.GlobalConfiguration - Loading configuration property: taskmanager.heap.size, 1024m
2020-06-01 14:43:42,022 INFO org.apache.flink.configuration.GlobalConfiguration - Loading configuration property: taskmanager.numberOfTaskSlots, 1
2020-06-01 14:43:42,022 INFO org.apache.flink.configuration.GlobalConfiguration - Loading configuration property: parallelism.default, 1
2020-06-01 14:43:42,023 INFO org.apache.flink.configuration.GlobalConfiguration - Loading configuration property: jobmanager.execution.failover-strategy, region
2020-06-01 14:43:42,023 INFO org.apache.flink.configuration.GlobalConfiguration - Loading configuration property: jobmanager.archive.fs.dir, hdfs://dap/tmp/completed-jobs/
2020-06-01 14:43:42,023 INFO org.apache.flink.configuration.GlobalConfiguration - Loading configuration property: historyserver.web.address, 0.0.0.0
2020-06-01 14:43:42,023 INFO org.apache.flink.configuration.GlobalConfiguration - Loading configuration property: historyserver.web.port, 8082
2020-06-01 14:43:42,023 INFO org.apache.flink.configuration.GlobalConfiguration - Loading configuration property: historyserver.archive.fs.dir, hdfs://dap/tmp/completed-jobs/
2020-06-01 14:43:42,023 INFO org.apache.flink.configuration.GlobalConfiguration - Loading configuration property: historyserver.archive.fs.refresh-interval, 10000
2020-06-01 14:43:42,069 INFO org.apache.flink.client.program.rest.RestClusterClient - Submitting job 410508f08b0775c0529e84b221dd909d (detached: false).
2020-06-01 14:43:52,134 INFO org.apache.flink.api.java.ExecutionEnvironment - The job has 0 registered types and 0 default Kryo serializers
2020-06-01 14:43:52,167 INFO org.apache.flink.optimizer.Optimizer -Compiler could not determine the size of input 'TextInputFormat ([hdfs://arc/success_fid_flow]) - UTF-8'. Using default estimates.
2020-06-01 14:43:52,171 INFO org.apache.flink.client.program.rest.RestClusterClient - Submitting job 10ac036dbb1d546996f2db76a2901253 (detached: false).
2020-06-01 14:43:55,649 INFO org.apache.flink.client.cli.CliFrontend - Program execution finished
Program execution finished
Job with JobID 10ac036dbb1d546996f2db76a2901253 has finished.
Job Runtime: 1143 ms
2020-06-01 14:43:55,663 INFO org.apache.flink.runtime.rest.RestClient - Shutting down rest endpoint.
2020-06-01 14:43:55,665 INFO org.apache.flink.runtime.rest.RestClient - Rest endpoint shutdown complete.
| |
王志华
|
|
a15733178518@163.com
|
签名由网易邮箱大师定制
Re: FileInputFormat 使用问题
Posted by john <50...@qq.com>.
嗨，找到问题了吗？我也遇到了
> 2020年6月1日 下午2:48，阿华田 <a1...@163.com> 写道：
>
> //初始化任务参数
> ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
> FileInputFormat fileInputFormat = new TextInputFormat(new Path("hdfs://arc/success_fid_flow <hdfs://arc/success_fid_flow>"));
> fileInputFormat.setNestedFileEnumeration(true);
> //过滤掉条件为true
> fileInputFormat.setFilesFilter(new RegexExcludePathAndTimeFilter("2020-05-24","2020-05-24"));
> DataSet<String> source =env.createInput(fileInputFormat);
> source.output(new HdfsTrainSinktest());
回复： FileInputFormat 使用问题
Posted by 阿华田 <a1...@163.com>.
解决了
| |
阿华田
|
|
a15733178518@163.com
|
签名由网易邮箱大师定制
在2020年06月18日 17:56，john<50...@qq.com> 写道：
嗨，找到问题了吗？我也遇到了
2020年6月1日 下午2:48，阿华田 <a1...@163.com> 写道：
//初始化任务参数
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
FileInputFormat fileInputFormat = new TextInputFormat(new Path("hdfs://arc/success_fid_flow <hdfs://arc/success_fid_flow>"));
fileInputFormat.setNestedFileEnumeration(true);
//过滤掉条件为true
fileInputFormat.setFilesFilter(new RegexExcludePathAndTimeFilter("2020-05-24","2020-05-24"));
DataSet<String> source =env.createInput(fileInputFormat);
source.output(new HdfsTrainSinktest());