Apache Flink 批处理模式在几分钟后失败并打印结果

Question

我正在使用 Apache Flink 读取 CSV 文件，然后将记录转换为我执行 SQL 查询并将结果打印到标准输出的表。

代码（简化）：

 StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(1);
 env.setRuntimeMode(RuntimeExecutionMode.BATCH);

 rowDataStreamSource = env.readFile(...).disableChaining();

 final StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
 final Table table = tableEnv.fromChangelogStream(rowDataStreamSource, ordersSchema, ChangelogMode.insertOnly());
 tableEnv.createTemporaryView("orders", table);

 Table originalSQL = tableEnv.sqlQuery(...)
 originalSQL.execute().print();

 env.execute();

运行此作业后几分钟后出现错误：

21:29:49.999 [deploy.instance_IS_UNDEFINED,,,][mini-cluster-io-thread-3] 信息 o.a.f.r.rpc.akka.AkkaRpcService - 停止 Akka RPC 服务。 21:29:50.066 [deploy.instance_IS_UNDEFINED,,,][http-nio-8080-exec-1] 警告 o.a.f.s.a.o.c.CollectResultFetcher - 无法获取作业状态，所以 我们假设工作已经终止。一些数据可能会丢失。 java.lang.IllegalStateException: MiniCluster 尚未运行或已 已经被关闭了。在 org.apache.flink.util.Preconditions.checkState(Preconditions.java:193) 在 org.apache.flink.runtime.minicluster.MiniCluster.getDispatcherGatewayFuture(MiniCluster.java:877)

然后在异常下方打印带有我的 sql 结果的表。

是什么导致了这里的问题？有没有办法告诉 Flink 没有更多记录，以便它可以完成工作并打印结果？

Answer 1

您应该使用FileSource 而不是readFile 以便在批处理执行模式下正常工作：https://nightlies.apache.org/flink/flink-docs-stable/api/java/org/apache/flink/connector/file/src/FileSource.html

或者，更好的是，您可以直接使用 SQL 定义一个表作为获取输入文件的源，如下所述：https://nightlies.apache.org/flink/flink-docs-stable/docs/connectors/table/filesystem/