Flink CsvTableSource Streaming

Question

我想流式传输一个csv文件并使用flink执行sql操作。但是我写的代码只读取一次并停止。它不流式传输。提前感谢，

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tableEnv = StreamTableEnvironment.getTableEnvironment(env);
CsvTableSource csvtable = CsvTableSource.builder()
.path("D:/employee.csv")
.ignoreFirstLine()
.fieldDelimiter(",")
.field("id", Types.INT())
.field("name", Types.STRING())
.field("designation", Types.STRING())
.field("age", Types.INT())
.field("location", Types.STRING())
.build();
tableEnv.registerTableSource("employee", csvtable);
Table table = tableEnv.scan("employee").where("name='jay'").select("id,name,location");
//Table table1 = tableEnv.scan("employee").where("age > 23").select("id,name,age,location");
DataStream<Row> stream = tableEnv.toAppendStream(table, Row.class);
//DataStream<Row> stream1 = tableEnv.toAppendStream(table1, Row.class);
stream.print();
//stream1.print();
env.execute();

Answer 1

CsvTableSource

基于逐行读取和解析引用文件FileInputFormat。生成的行将转发到流式处理查询中。因此，在CsvTableSource，流式传输是指行被连续读取和转发。但是，CsvTableSource在文件末尾终止。因此，它会发出有界流。

我假设您期望的行为是CsvTableSource读取文件直到其结束，然后等待将写入追加到文件。 但是，这不是CsvTableSource的工作方式。您需要为此实现自定义TableSource。